jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in feature/localstackands3object (pull request #54)

Browse Source 
LocalStack and S3 Objects

* firstround

* cleanupintegration

* tidyports
This commit is contained in:
Michael McGuinness
2025-02-06 15:00:26 +00:00
parent 9e6328d8f9
commit e146725825
17 changed files with 381 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/document/init/create.go
+28 -17
View File
@@ -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
}
internal/document/init/create_test.go
+21 -49
View File
@@ -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)
}
internal/document/init/service.go
-2
View File
@@ -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 {
internal/server/runner/poll.go
+3 -3
View File
@@ -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)
}
}
internal/serviceconfig/objectstore/client.go
+4
View File
@@ -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)
}