diff --git a/.gitignore b/.gitignore index 8a7669cb..8aef8be5 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,6 @@ +**/*.http +**/*.cast + # Ignore the binary files generated by `go build` /main diff --git a/api/docInitRunner/runner.go b/api/docInitRunner/runner.go index 12715e5e..5c6a79d2 100644 --- a/api/docInitRunner/runner.go +++ b/api/docInitRunner/runner.go @@ -3,6 +3,9 @@ package docinitrunner import ( "context" "encoding/json" + "errors" + "fmt" + "log/slog" documentinit "queryorchestration/internal/document/init" "github.com/go-playground/validator/v10" @@ -28,8 +31,30 @@ func New(validator *validator.Validate, svc *Services) Runner { } } +type S3EventNotification struct { + Records []struct { + EventVersion string `json:"eventVersion"` + EventSource string `json:"eventSource"` + AwsRegion string `json:"awsRegion"` + EventTime string `json:"eventTime"` + EventName string `json:"eventName"` + S3 struct { + Bucket struct { + Name string `json:"name"` + Arn string `json:"arn"` + } `json:"bucket"` + Object struct { + Key string `json:"key"` + Size int64 `json:"size"` + ETag string `json:"eTag"` + VersionId string `json:"versionId"` + } `json:"object"` + } `json:"s3"` + } `json:"Records"` +} + func (s Runner) Process(ctx context.Context, req *types.Message) error { - var body documentinit.Create + var body S3EventNotification err := json.Unmarshal([]byte(*req.Body), &body) if err != nil { return err @@ -40,9 +65,39 @@ func (s Runner) Process(ctx context.Context, req *types.Message) error { return err } - _, err = s.svc.Document.Create(ctx, &body) - if err != nil { - return err + 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: record.S3.Object.Key, + Bucket: record.S3.Bucket.Name, + Hash: record.S3.Object.ETag, + }) + if err != nil { + return err + } + + return nil + }() + if err != nil { + hasErr = true + slog.Error("Error processing record", "err", err) + } + } + + if hasErr { + return errors.New("error processing records") } return nil diff --git a/api/docInitRunner/runner_test.go b/api/docInitRunner/runner_test.go index 234808da..1b1ea896 100644 --- a/api/docInitRunner/runner_test.go +++ b/api/docInitRunner/runner_test.go @@ -17,13 +17,11 @@ import ( 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 { @@ -98,18 +96,6 @@ func TestDocInitRunner(t *testing.T) { 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/cmd/docInitRunner/main.go b/cmd/docInitRunner/main.go index cd40d29d..9d5d7480 100644 --- a/cmd/docInitRunner/main.go +++ b/cmd/docInitRunner/main.go @@ -13,7 +13,6 @@ import ( "queryorchestration/internal/job/collector" "queryorchestration/internal/server/runner" "queryorchestration/internal/serviceconfig" - objectstore "queryorchestration/internal/serviceconfig/objectstore" documentcleanc "queryorchestration/internal/serviceconfig/queue/documentclean" _ "github.com/lib/pq" @@ -22,7 +21,6 @@ import ( type DocInitConfig struct { runner.BaseConfig documentcleanc.DocCleanConfig - objectstore.ObjectStoreConfig } func main() { @@ -62,11 +60,5 @@ func main() { os.Exit(1) } - err = cfg.SetStoreClient(ctx) - if err != nil { - slog.Error(err.Error()) - os.Exit(1) - } - server.Listen(ctx) } diff --git a/deployments/compose.local.yaml b/deployments/compose.local.yaml index 7997efbc..944daef2 100644 --- a/deployments/compose.local.yaml +++ b/deployments/compose.local.yaml @@ -1,5 +1,75 @@ --- services: + doc_init_runner: + image: queryorchestration:latest + command: ["./docInitRunner"] + depends_on: + - db + - localstack + environment: + QUEUE_URL: ${DOCUMENT_INIT_URL} + DOCUMENT_CLEAN_URL: ${DOCUMENT_CLEAN_URL} + AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID} + AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY} + AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN} + AWS_REGION: ${AWS_REGION} + DB_USER: ${DB_USER} + DB_PASS: ${DB_PASS} + DB_HOST: db + DB_PORT: 5432 + DB_NAME: ${DB_NAME} + DB_NOSSL: ${DB_NOSSL} + AWS_ENDPOINT_URL: "http://localstack:4566" + AWS_S3_USE_PATH_STYLE: true + networks: + - server-network + query_runner: + image: queryorchestration:latest + command: ["./queryRunner"] + depends_on: + - db + - localstack + environment: + QUEUE_URL: ${QUERY_RUNNER_URL} + AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID} + AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY} + AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN} + AWS_REGION: ${AWS_REGION} + DB_USER: ${DB_USER} + DB_PASS: ${DB_PASS} + DB_HOST: db + DB_PORT: 5432 + DB_NAME: ${DB_NAME} + DB_NOSSL: ${DB_NOSSL} + AWS_ENDPOINT_URL: "http://localstack:4566" + AWS_S3_USE_PATH_STYLE: true + networks: + - server-network + query_service: + image: queryorchestration:latest + command: ["./queryService"] + depends_on: + - db + - localstack + ports: + - 8080:8080 + expose: + - 8080 + environment: + AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID} + AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY} + AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN} + AWS_REGION: ${AWS_REGION} + DB_USER: ${DB_USER} + DB_PASS: ${DB_PASS} + DB_HOST: db + DB_PORT: 5432 + DB_NAME: ${DB_NAME} + DB_NOSSL: ${DB_NOSSL} + AWS_ENDPOINT_URL: "http://localstack:4566" + AWS_S3_USE_PATH_STYLE: true + networks: + - server-network db: image: postgres:17.2-alpine3.21 ports: diff --git a/deployments/compose.test.yaml b/deployments/compose.test.yaml index 416adbb4..69343d10 100644 --- a/deployments/compose.test.yaml +++ b/deployments/compose.test.yaml @@ -3,13 +3,13 @@ services: test-db: image: postgres:17.2-alpine3.21 ports: - - 5432:5432 + - 5430:5432 environment: POSTGRES_DB: ${DB_NAME} POSTGRES_PASSWORD: ${DB_PASS} POSTGRES_USER: ${DB_USER} expose: - - 5432 + - 5430 volumes: - test-db-data:/var/lib/postgresql/data healthcheck: diff --git a/devbox.json b/devbox.json index 28d1cd1a..b134e999 100644 --- a/devbox.json +++ b/devbox.json @@ -22,6 +22,7 @@ "shell": { "init_hook": [ "export DB_URI=postgres://${DB_USER}:${DB_PASS}@${DB_HOST}:${DB_PORT}/${DB_NAME}?sslmode=disable", + "export DB_URI_TEST=postgres://${DB_USER}:${DB_PASS}@${DB_HOST}:${DB_PORT_TEST}/${DB_NAME}?sslmode=disable", "echo 'Welcome to the DoczyAI devbox!'" ] }, @@ -31,6 +32,7 @@ "DB_PASS": "pass", "DB_HOST": "localhost", "DB_PORT": "5432", + "DB_PORT_TEST": "5430", "DB_NAME": "query_orchestration", "DB_NOSSL": "true", "AWS_ACCESS_KEY_ID": "test", diff --git a/internal/document/init/create.go b/internal/document/init/create.go index b7044a6d..9dab210c 100644 --- a/internal/document/init/create.go +++ b/internal/document/init/create.go @@ -4,23 +4,25 @@ import ( "context" "database/sql" "errors" + "fmt" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" documentclean "queryorchestration/internal/document/clean" "queryorchestration/internal/job" "queryorchestration/internal/serviceconfig/queue" + "regexp" - "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"` + JobID uuid.UUID + Location document.Location + Bucket string + Hash string } func (s *Service) Create(ctx context.Context, doc *Create) (uuid.UUID, error) { @@ -39,9 +41,11 @@ func (s *Service) Create(ctx context.Context, doc *Create) (uuid.UUID, error) { return uuid.Nil, err } - err = s.informCreate(ctx, id, j) - if err != nil { - return uuid.Nil, err + if params.ID == nil { + err = s.informCreate(ctx, id, j) + if err != nil { + return uuid.Nil, err + } } return id, nil @@ -56,17 +60,9 @@ type createDocumentParams struct { } 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 - } - idbyhash, err := s.cfg.GetDBQueries().GetDocumentIDByHash(ctx, &repository.GetDocumentIDByHashParams{ Jobid: database.MustToDBUUID(doc.JobID), - Hash: *res.ETag, + Hash: doc.Hash, }) var docID *pgtype.UUID if err != nil && !errors.Is(err, sql.ErrNoRows) { @@ -78,7 +74,7 @@ func (s *Service) getCreateParams(ctx context.Context, doc *Create) (*createDocu return &createDocumentParams{ ID: docID, JobID: database.MustToDBUUID(doc.JobID), - Hash: *res.ETag, + Hash: doc.Hash, Bucket: doc.Bucket, Location: doc.Location, }, nil @@ -141,3 +137,18 @@ func (s *Service) informCreate(ctx context.Context, id uuid.UUID, j *job.Job) er return nil } + +func (s *Service) GetJobIDFromKey(key string) (uuid.UUID, error) { + re := regexp.MustCompile(`^.+/(.+)/.+$`) + match := re.FindStringSubmatch(key) + if match == nil || len(match) != 2 { + return uuid.Nil, fmt.Errorf("no job id match found in key") + } + + id, err := uuid.Parse(match[1]) + if err != nil { + return uuid.Nil, err + } + + return id, nil +} diff --git a/internal/document/init/create_test.go b/internal/document/init/create_test.go index 8453f21b..f296e4f4 100644 --- a/internal/document/init/create_test.go +++ b/internal/document/init/create_test.go @@ -16,7 +16,6 @@ import ( 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" @@ -82,22 +81,11 @@ func TestCreate(t *testing.T) { 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: location, Bucket: bucket, + Hash: doc.Hash, }) assert.NoError(t, err) assert.Equal(t, doc.ID, id) @@ -164,22 +152,11 @@ func TestGetCreateParams(t *testing.T) { 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, + Hash: doc.Hash, }) assert.NoError(t, err) assert.Equal(t, &createDocumentParams{ @@ -222,22 +199,11 @@ func TestGetCreateParamsExisting(t *testing.T) { 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, + Hash: doc.Hash, }) assert.NoError(t, err) dbid := database.MustToDBUUID(doc.ID) @@ -279,18 +245,6 @@ func TestGetCreateParamsCurrentDocErr(t *testing.T) { 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, @@ -385,3 +339,21 @@ func TestSubmitCreateExists(t *testing.T) { assert.NoError(t, err) assert.Equal(t, doc.ID, id) } + +func TestGetJobIDFromKey(t *testing.T) { + svc := Service{} + + _, err := svc.GetJobIDFromKey("") + assert.Error(t, err) + + _, err = svc.GetJobIDFromKey("jkhaskhweuifhwhieh") + assert.Error(t, err) + + _, err = svc.GetJobIDFromKey("aaa/bbb/ccc") + assert.Error(t, err) + + id := uuid.New() + aid, err := svc.GetJobIDFromKey(fmt.Sprintf("aaa/%s/bbb", id.String())) + assert.NoError(t, err) + assert.Equal(t, id, aid) +} diff --git a/internal/document/init/service.go b/internal/document/init/service.go index 66090395..cb62ad90 100644 --- a/internal/document/init/service.go +++ b/internal/document/init/service.go @@ -3,7 +3,6 @@ package documentinit import ( "queryorchestration/internal/job" "queryorchestration/internal/serviceconfig" - "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/documentclean" ) @@ -14,7 +13,6 @@ type Services struct { type ConfigProvider interface { serviceconfig.ConfigProvider documentclean.ConfigProvider - objectstore.ConfigProvider } type Service struct { diff --git a/internal/server/runner/poll.go b/internal/server/runner/poll.go index d64a1e1c..92b237cd 100644 --- a/internal/server/runner/poll.go +++ b/internal/server/runner/poll.go @@ -43,11 +43,11 @@ func (c *Server) pollMessage(ctx context.Context) error { } for _, message := range result.Messages { - slog.Info("processing message", "id", message.MessageId) + slog.Info("processing message", "id", *message.MessageId) err := c.cfg.GetController().Process(ctx, &message) if err != nil { - slog.Error("message process fail", "err", err) + slog.Error("message process fail", "id", *message.MessageId, "err", err) } err = c.cfg.DeleteFromQueue(ctx, &queue.DeleteParams{ @@ -55,7 +55,7 @@ func (c *Server) pollMessage(ctx context.Context) error { ReceiptHandle: message.ReceiptHandle, }) if err != nil { - slog.Error("message delete fail", "err", err) + slog.Error("message delete fail", "id", *message.MessageId, "err", err) } } diff --git a/internal/serviceconfig/objectstore/client.go b/internal/serviceconfig/objectstore/client.go index c14903c7..d73632c9 100644 --- a/internal/serviceconfig/objectstore/client.go +++ b/internal/serviceconfig/objectstore/client.go @@ -47,4 +47,8 @@ type S3Client interface { GetBucketLifecycleConfiguration(ctx context.Context, params *s3.GetBucketLifecycleConfigurationInput, opts ...func(*s3.Options)) (*s3.GetBucketLifecycleConfigurationOutput, error) PutBucketLifecycleConfiguration(ctx context.Context, params *s3.PutBucketLifecycleConfigurationInput, opts ...func(*s3.Options)) (*s3.PutBucketLifecycleConfigurationOutput, error) DeleteBucketLifecycle(ctx context.Context, params *s3.DeleteBucketLifecycleInput, opts ...func(*s3.Options)) (*s3.DeleteBucketLifecycleOutput, error) + + // Bucket Notifications + PutBucketNotificationConfiguration(ctx context.Context, params *s3.PutBucketNotificationConfigurationInput, opts ...func(*s3.Options)) (*s3.PutBucketNotificationConfigurationOutput, error) + GetBucketNotificationConfiguration(ctx context.Context, params *s3.GetBucketNotificationConfigurationInput, opts ...func(*s3.Options)) (*s3.GetBucketNotificationConfigurationOutput, error) } diff --git a/mocks/objectstore/mock_S3Client.go b/mocks/objectstore/mock_S3Client.go index d697ba37..4dd4f295 100644 --- a/mocks/objectstore/mock_S3Client.go +++ b/mocks/objectstore/mock_S3Client.go @@ -911,6 +911,80 @@ func (_c *MockS3Client_GetBucketLifecycleConfiguration_Call) RunAndReturn(run fu return _c } +// GetBucketNotificationConfiguration provides a mock function with given fields: ctx, params, opts +func (_m *MockS3Client) GetBucketNotificationConfiguration(ctx context.Context, params *s3.GetBucketNotificationConfigurationInput, opts ...func(*s3.Options)) (*s3.GetBucketNotificationConfigurationOutput, error) { + _va := make([]interface{}, len(opts)) + for _i := range opts { + _va[_i] = opts[_i] + } + var _ca []interface{} + _ca = append(_ca, ctx, params) + _ca = append(_ca, _va...) + ret := _m.Called(_ca...) + + if len(ret) == 0 { + panic("no return value specified for GetBucketNotificationConfiguration") + } + + var r0 *s3.GetBucketNotificationConfigurationOutput + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, *s3.GetBucketNotificationConfigurationInput, ...func(*s3.Options)) (*s3.GetBucketNotificationConfigurationOutput, error)); ok { + return rf(ctx, params, opts...) + } + if rf, ok := ret.Get(0).(func(context.Context, *s3.GetBucketNotificationConfigurationInput, ...func(*s3.Options)) *s3.GetBucketNotificationConfigurationOutput); ok { + r0 = rf(ctx, params, opts...) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*s3.GetBucketNotificationConfigurationOutput) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, *s3.GetBucketNotificationConfigurationInput, ...func(*s3.Options)) error); ok { + r1 = rf(ctx, params, opts...) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// MockS3Client_GetBucketNotificationConfiguration_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetBucketNotificationConfiguration' +type MockS3Client_GetBucketNotificationConfiguration_Call struct { + *mock.Call +} + +// GetBucketNotificationConfiguration is a helper method to define mock.On call +// - ctx context.Context +// - params *s3.GetBucketNotificationConfigurationInput +// - opts ...func(*s3.Options) +func (_e *MockS3Client_Expecter) GetBucketNotificationConfiguration(ctx interface{}, params interface{}, opts ...interface{}) *MockS3Client_GetBucketNotificationConfiguration_Call { + return &MockS3Client_GetBucketNotificationConfiguration_Call{Call: _e.mock.On("GetBucketNotificationConfiguration", + append([]interface{}{ctx, params}, opts...)...)} +} + +func (_c *MockS3Client_GetBucketNotificationConfiguration_Call) Run(run func(ctx context.Context, params *s3.GetBucketNotificationConfigurationInput, opts ...func(*s3.Options))) *MockS3Client_GetBucketNotificationConfiguration_Call { + _c.Call.Run(func(args mock.Arguments) { + variadicArgs := make([]func(*s3.Options), len(args)-2) + for i, a := range args[2:] { + if a != nil { + variadicArgs[i] = a.(func(*s3.Options)) + } + } + run(args[0].(context.Context), args[1].(*s3.GetBucketNotificationConfigurationInput), variadicArgs...) + }) + return _c +} + +func (_c *MockS3Client_GetBucketNotificationConfiguration_Call) Return(_a0 *s3.GetBucketNotificationConfigurationOutput, _a1 error) *MockS3Client_GetBucketNotificationConfiguration_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *MockS3Client_GetBucketNotificationConfiguration_Call) RunAndReturn(run func(context.Context, *s3.GetBucketNotificationConfigurationInput, ...func(*s3.Options)) (*s3.GetBucketNotificationConfigurationOutput, error)) *MockS3Client_GetBucketNotificationConfiguration_Call { + _c.Call.Return(run) + return _c +} + // GetObject provides a mock function with given fields: ctx, params, opts func (_m *MockS3Client) GetObject(ctx context.Context, params *s3.GetObjectInput, opts ...func(*s3.Options)) (*s3.GetObjectOutput, error) { _va := make([]interface{}, len(opts)) @@ -1799,6 +1873,80 @@ func (_c *MockS3Client_PutBucketLifecycleConfiguration_Call) RunAndReturn(run fu return _c } +// PutBucketNotificationConfiguration provides a mock function with given fields: ctx, params, opts +func (_m *MockS3Client) PutBucketNotificationConfiguration(ctx context.Context, params *s3.PutBucketNotificationConfigurationInput, opts ...func(*s3.Options)) (*s3.PutBucketNotificationConfigurationOutput, error) { + _va := make([]interface{}, len(opts)) + for _i := range opts { + _va[_i] = opts[_i] + } + var _ca []interface{} + _ca = append(_ca, ctx, params) + _ca = append(_ca, _va...) + ret := _m.Called(_ca...) + + if len(ret) == 0 { + panic("no return value specified for PutBucketNotificationConfiguration") + } + + var r0 *s3.PutBucketNotificationConfigurationOutput + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, *s3.PutBucketNotificationConfigurationInput, ...func(*s3.Options)) (*s3.PutBucketNotificationConfigurationOutput, error)); ok { + return rf(ctx, params, opts...) + } + if rf, ok := ret.Get(0).(func(context.Context, *s3.PutBucketNotificationConfigurationInput, ...func(*s3.Options)) *s3.PutBucketNotificationConfigurationOutput); ok { + r0 = rf(ctx, params, opts...) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*s3.PutBucketNotificationConfigurationOutput) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, *s3.PutBucketNotificationConfigurationInput, ...func(*s3.Options)) error); ok { + r1 = rf(ctx, params, opts...) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// MockS3Client_PutBucketNotificationConfiguration_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'PutBucketNotificationConfiguration' +type MockS3Client_PutBucketNotificationConfiguration_Call struct { + *mock.Call +} + +// PutBucketNotificationConfiguration is a helper method to define mock.On call +// - ctx context.Context +// - params *s3.PutBucketNotificationConfigurationInput +// - opts ...func(*s3.Options) +func (_e *MockS3Client_Expecter) PutBucketNotificationConfiguration(ctx interface{}, params interface{}, opts ...interface{}) *MockS3Client_PutBucketNotificationConfiguration_Call { + return &MockS3Client_PutBucketNotificationConfiguration_Call{Call: _e.mock.On("PutBucketNotificationConfiguration", + append([]interface{}{ctx, params}, opts...)...)} +} + +func (_c *MockS3Client_PutBucketNotificationConfiguration_Call) Run(run func(ctx context.Context, params *s3.PutBucketNotificationConfigurationInput, opts ...func(*s3.Options))) *MockS3Client_PutBucketNotificationConfiguration_Call { + _c.Call.Run(func(args mock.Arguments) { + variadicArgs := make([]func(*s3.Options), len(args)-2) + for i, a := range args[2:] { + if a != nil { + variadicArgs[i] = a.(func(*s3.Options)) + } + } + run(args[0].(context.Context), args[1].(*s3.PutBucketNotificationConfigurationInput), variadicArgs...) + }) + return _c +} + +func (_c *MockS3Client_PutBucketNotificationConfiguration_Call) Return(_a0 *s3.PutBucketNotificationConfigurationOutput, _a1 error) *MockS3Client_PutBucketNotificationConfiguration_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *MockS3Client_PutBucketNotificationConfiguration_Call) RunAndReturn(run func(context.Context, *s3.PutBucketNotificationConfigurationInput, ...func(*s3.Options)) (*s3.PutBucketNotificationConfigurationOutput, error)) *MockS3Client_PutBucketNotificationConfiguration_Call { + _c.Call.Return(run) + return _c +} + // PutObject provides a mock function with given fields: ctx, params, opts func (_m *MockS3Client) PutObject(ctx context.Context, params *s3.PutObjectInput, opts ...func(*s3.Options)) (*s3.PutObjectOutput, error) { _va := make([]interface{}, len(opts)) diff --git a/scripts/compose.yml b/scripts/compose.yml index 5d8af94f..9a31594a 100644 --- a/scripts/compose.yml +++ b/scripts/compose.yml @@ -21,16 +21,30 @@ tasks: up:test: cmds: - docker compose -f {{.TEST_COMPOSE_FILE}} up --no-recreate -d + refresh: + cmds: + - task: down + - task: up up: cmds: + - task docker:build - docker compose -f {{.LOCAL_COMPOSE_FILE}} up --no-recreate -d - task: init init: cmds: - - aws sqs create-queue --queue-name $QNAME_DOCUMENT_CLEAN - - aws sqs create-queue --queue-name $QNAME_DOCUMENT_INIT - - aws sqs create-queue --queue-name $QNAME_QUERY_RUNNER - aws s3 mb s3://$BUCKET_IN + - aws sqs create-queue --queue-name $QNAME_DOCUMENT_INIT + - | + aws s3api put-bucket-notification-configuration \ + --bucket $BUCKET_IN \ + --notification-configuration '{ + "QueueConfigurations": [{ + "QueueArn": "arn:aws:sqs:us-east-1:000000000000:document_init", + "Events": ["s3:ObjectCreated:*"] + }] + }' + - aws sqs create-queue --queue-name $QNAME_DOCUMENT_CLEAN + - aws sqs create-queue --queue-name $QNAME_QUERY_RUNNER down:test: cmds: - task: down @@ -51,15 +65,7 @@ tasks: - docker volume rm deployments_db-data lint: cmds: - - task: lint:file - vars: - COMPOSE_FILE: '{{.LOCAL_COMPOSE_FILE}}' - - task: lint:file - vars: - COMPOSE_FILE: '{{.TEST_COMPOSE_FILE}}' - lint:file: - internal: true - vars: - COMPOSE_FILE: '{{.COMPOSE_FILE | default .LOCAL_COMPOSE_FILE}}' - cmds: - - docker compose -f {{.COMPOSE_FILE}} config -q --resolve-image-digests + - docker compose -f {{.LOCAL_COMPOSE_FILE}} config -q + - | + docker compose -f {{.TEST_COMPOSE_FILE}} config -q \ + --resolve-image-digests diff --git a/scripts/database.yml b/scripts/database.yml index 65f5eaaa..fb98d3fc 100644 --- a/scripts/database.yml +++ b/scripts/database.yml @@ -32,4 +32,4 @@ tasks: deps: - deps:up cmds: - - migrate -path {{.MIGRATIONS}} -database {{.DB_URI}} up + - migrate -path {{.MIGRATIONS}} -database {{.DB_URI_TEST}} up diff --git a/sqlc.yml b/sqlc.yml index bffbe779..e1a6712e 100644 --- a/sqlc.yml +++ b/sqlc.yml @@ -2,7 +2,7 @@ version: "2" servers: - engine: postgresql - uri: "${DB_URI}" + uri: "${DB_URI_TEST}" sql: - name: "db" engine: "postgresql" diff --git a/test/docInitRunner/docinitrunner_test.go b/test/docInitRunner/docinitrunner_test.go index 11fb6237..b931aa13 100644 --- a/test/docInitRunner/docinitrunner_test.go +++ b/test/docInitRunner/docinitrunner_test.go @@ -2,12 +2,11 @@ package integration_test import ( "context" + "fmt" "os" "path" - documentinit "queryorchestration/internal/document/init" "queryorchestration/internal/server/runner" "queryorchestration/internal/serviceconfig/objectstore" - "queryorchestration/internal/serviceconfig/queue" documentcleanc "queryorchestration/internal/serviceconfig/queue/documentclean" "queryorchestration/internal/test" queryservice "queryorchestration/pkg/queryService" @@ -16,6 +15,7 @@ import ( "testing" "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" "github.com/stretchr/testify/assert" ) @@ -49,6 +49,21 @@ func TestDocInitRunner(t *testing.T) { doccleanurl := test.CreateQueue(t, ctx, cfg, "docclean") bucketName := "docinitbucket" test.CreateBucket(t, ctx, cfg, bucketName) + arn := fmt.Sprintf("arn:aws:sqs:%s:000000000000:%s", cfg.AWSRegion, test.DocInitRunner) + _, err = cfg.StoreClient.PutBucketNotificationConfiguration(ctx, &s3.PutBucketNotificationConfigurationInput{ + Bucket: &bucketName, + NotificationConfiguration: &types.NotificationConfiguration{ + QueueConfigurations: []types.QueueConfiguration{ + { + QueueArn: &arn, + Events: []types.Event{ + types.EventS3ObjectCreated, + }, + }, + }, + }, + }) + assert.NoError(t, err) net, cleanup := test.CreateRunnersAndServicesNetwork(t, ctx, &test.EcosystemNetworkConfig{ Cfg: cfg, @@ -91,7 +106,7 @@ func TestDocInitRunner(t *testing.T) { }) assert.NoError(t, err) - location := "/i/am/here" + location := fmt.Sprintf("%s/%s/%s", clientRes.JSON201.Id, jobRes.JSON201.Id, "object_name") body := strings.NewReader("hello world") _, err = cfg.StoreClient.PutObject(ctx, &s3.PutObjectInput{ Bucket: &bucketName, @@ -100,15 +115,6 @@ func TestDocInitRunner(t *testing.T) { }) assert.NoError(t, err) - 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) }