Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(server,worker): resolve copier metadata bug #1364

Merged
merged 6 commits into from
Jan 20, 2025
Merged

fix(server,worker): resolve copier metadata bug #1364

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
bda0fae
use reearth cms db
nourbalaha Jan 20, 2025
08d29fe
wip
nourbalaha Jan 20, 2025
3d30be4
wip
nourbalaha Jan 20, 2025
86dba9b
fix: TestModel_Copy
nourbalaha Jan 20, 2025
78eb1ca
error handling
nourbalaha Jan 20, 2025
996f1c4
use util.now
nourbalaha Jan 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion server/internal/infrastructure/gcp/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,5 +12,5 @@ type TaskConfig struct {
DecompressorMachineType string `default:"E2_HIGHCPU_8"`
DecompressorDiskSideGb int64 `default:"2000"`
CopierImage string `default:"reearth/reearth-cms-copier"`
DBSecretName string `default:"reearth-cms-worker-db"`
DBSecretName string `default:"reearth-cms-db"`
}
4 changes: 2 additions & 2 deletions server/internal/infrastructure/gcp/taskrunner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ func copy(ctx context.Context, p task.Payload, conf *TaskConfig) error {
"REEARTH_CMS_COPIER_CHANGES=" + p.Copy.Changes,
},
SecretEnv: []string{
"REEARTH_CMS_WORKER_DB",
"REEARTH_CMS_DB",
},
},
},
Expand All @@ -181,7 +181,7 @@ func copy(ctx context.Context, p task.Payload, conf *TaskConfig) error {
SecretManager: []*cloudbuild.SecretManagerSecret{
{
VersionName: fmt.Sprintf("projects/%s/secrets/%s/versions/latest", project, dbSecretName),
Env: "REEARTH_CMS_WORKER_DB",
Env: "REEARTH_CMS_DB",
},
},
},
Expand Down
63 changes: 55 additions & 8 deletions server/internal/usecase/interactor/model.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"fmt"
"time"

"github.com/reearth/reearth-cms/server/internal/usecase"
"github.com/reearth/reearth-cms/server/internal/usecase/gateway"
Expand Down Expand Up @@ -372,7 +373,8 @@ func (i Model) Copy(ctx context.Context, params interfaces.CopyModelParam, opera
}

// Copy items
if err := i.copyItems(ctx, oldModel.Schema(), newModel.Schema(), newModel.ID()); err != nil {
timestamp := time.Now().UnixMilli()
if err := i.copyItems(ctx, oldModel.Schema(), newModel.Schema(), newModel.ID(), timestamp, operator); err != nil {
nourbalaha marked this conversation as resolved.
Show resolved Hide resolved
return nil, err
}

Expand Down Expand Up @@ -403,7 +405,7 @@ func (i Model) Copy(ctx context.Context, params interfaces.CopyModelParam, opera
return nil, err
}

if err := i.copyItems(ctx, *oldModel.Metadata(), newMetaSchema.ID(), newModel.ID()); err != nil {
if err := i.copyItems(ctx, *oldModel.Metadata(), newMetaSchema.ID(), newModel.ID(), timestamp, operator); err != nil {
return nil, err
}
}
Expand All @@ -413,16 +415,16 @@ func (i Model) Copy(ctx context.Context, params interfaces.CopyModelParam, opera
})
}

func (i Model) copyItems(ctx context.Context, oldSchemaID, newSchemaID id.SchemaID, newModelID id.ModelID) error {
func (i Model) copyItems(ctx context.Context, oldSchemaID, newSchemaID id.SchemaID, newModelID id.ModelID, timestamp int64, operator *usecase.Operator) error {
collection := "item"
filter, err := json.Marshal(bson.M{"schema": oldSchemaID.String()})
filter, err := json.Marshal(bson.M{"schema": oldSchemaID.String(), "__r": bson.M{"$in": []string{"latest"}}})
if err != nil {
return err
}
changes, err := json.Marshal(task.Changes{
c := task.Changes{
"id": {
Type: task.ChangeTypeNew,
Value: "item",
Type: task.ChangeTypeULID,
Value: timestamp,
},
"schema": {
Type: task.ChangeTypeSet,
Expand All @@ -432,7 +434,52 @@ func (i Model) copyItems(ctx context.Context, oldSchemaID, newSchemaID id.Schema
Type: task.ChangeTypeSet,
Value: newModelID.String(),
},
})
"timestamp": {
Type: task.ChangeTypeSet,
Value: time.Now().String(),
},
nourbalaha marked this conversation as resolved.
Show resolved Hide resolved
"updatedbyuser": {
Type: task.ChangeTypeSet,
Value: nil,
},
"updatedbyintegration": {
Type: task.ChangeTypeSet,
Value: nil,
},
"originalitem": {
Type: task.ChangeTypeULID,
Value: timestamp,
},
"metadataitem": {
Type: task.ChangeTypeULID,
Value: timestamp,
},
"__r": { // tag
Type: task.ChangeTypeSet,
Value: []string{"latest"},
},
"__w": { // parent
Type: task.ChangeTypeSet,
Value: nil,
},
"__v": { // version
Type: task.ChangeTypeNew,
Value: "version",
},
}
if operator.AcOperator.User != nil {
c["user"] = task.Change{
Type: task.ChangeTypeSet,
Value: operator.AcOperator.User.String(),
}
}
nourbalaha marked this conversation as resolved.
Show resolved Hide resolved
if operator.Integration != nil {
c["integration"] = task.Change{
Type: task.ChangeTypeSet,
Value: operator.Integration.String(),
}
}
nourbalaha marked this conversation as resolved.
Show resolved Hide resolved
changes, err := json.Marshal(c)
if err != nil {
return err
}
Expand Down
7 changes: 6 additions & 1 deletion server/internal/usecase/interactor/model_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -541,7 +541,12 @@ func TestModel_Copy(t *testing.T) {
mockTime := time.Now()
wid := accountdomain.NewWorkspaceID()
p := project.New().NewID().Workspace(wid).MustBuild()
op := &usecase.Operator{OwningProjects: []id.ProjectID{p.ID()}}
op := &usecase.Operator{
OwningProjects: []id.ProjectID{p.ID()},
AcOperator: &accountusecase.Operator{
User: accountdomain.NewUserID().Ref(),
},
}

fId1 := id.NewFieldID()
sfKey1 := id.RandomKey()
Expand Down
7 changes: 4 additions & 3 deletions server/pkg/task/task.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,11 +64,12 @@ func (p *CopyPayload) Validate() bool {
type Changes map[string]Change
type Change struct {
Type ChangeType
Value string
Value any
}
type ChangeType string

const (
ChangeTypeSet ChangeType = "set"
ChangeTypeNew ChangeType = "new"
ChangeTypeSet ChangeType = "set"
ChangeTypeNew ChangeType = "new"
ChangeTypeULID ChangeType = "ulid"
)
2 changes: 1 addition & 1 deletion worker/cmd/copier/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ func main() {
log.Info("config: .env loaded")
}

dbURI := mustGetEnv("REEARTH_CMS_WORKER_DB")
dbURI := mustGetEnv("REEARTH_CMS_DB")
collection := mustGetEnv("REEARTH_CMS_COPIER_COLLECTION")
filter := mustGetEnv("REEARTH_CMS_COPIER_FILTER")
changes := mustGetEnv("REEARTH_CMS_COPIER_CHANGES")
Expand Down
27 changes: 26 additions & 1 deletion worker/internal/infrastructure/mongo/copier.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@ import (
"context"
"errors"

"github.com/google/uuid"
"github.com/oklog/ulid"
"github.com/reearth/reearth-cms/server/pkg/id"
"github.com/reearth/reearth-cms/server/pkg/task"
"github.com/reearth/reearth-cms/worker/internal/usecase/repo"
Expand Down Expand Up @@ -58,7 +60,28 @@ func (r *Copier) Copy(ctx context.Context, f bson.M, changesMap task.Changes) er
for k, change := range changesMap {
switch change.Type {
case task.ChangeTypeNew:
newId, _ := generateId(change.Value)
str, ok := change.Value.(string)
if !ok {
return rerror.ErrInternalBy(err)
}
newId, ok := generateId(str)
if !ok {
return rerror.ErrInternalBy(err)
}
result[k] = newId
nourbalaha marked this conversation as resolved.
Show resolved Hide resolved
case task.ChangeTypeULID:
newId, err := ulid.Parse(result[k].(string))
if err != nil {
return rerror.ErrInternalBy(err)
}
v, ok := change.Value.(uint64)
if !ok {
return rerror.ErrInternalBy(err)
}
err = newId.SetTime(v)
if err != nil {
return rerror.ErrInternalBy(err)
}
result[k] = newId
case task.ChangeTypeSet:
result[k] = change.Value
Expand Down Expand Up @@ -91,6 +114,8 @@ func generateId(t string) (string, bool) {
return id.NewSchemaID().String(), true
case "model":
return id.NewModelID().String(), true
case "version":
return uuid.New().String(), true
default:
return "", false
}
Expand Down
Loading