feat(server): move export csv items to usecase layer

server/internal/adapter/integration/item.go

@@ -103,26 +103,14 @@ func (s *Server) ItemsAsCSV(ctx context.Context, request ItemsAsCSVRequestObject
 	op := adapter.Operator(ctx)
 	uc := adapter.Usecases(ctx)
 
-	sp, err := uc.Schema.FindByModel(ctx, request.ModelId, op)
-	if err != nil {
-		return ItemsAsCSV400Response{}, err
-	}
-
-	p := fromPagination(request.Params.Page, request.Params.PerPage)
-	items, _, err := uc.Item.FindBySchema(ctx, sp.Schema().ID(), nil, p, op)
+	pr, err := uc.Item.ItemsAsCSV(ctx, request.ModelId, request.Params.Page, request.Params.PerPage, op)
 	if err != nil {
 		if errors.Is(err, rerror.ErrNotFound) {
 			return ItemsAsCSV404Response{}, err
 		}
 		return ItemsAsCSV400Response{}, err
 	}
 
-	pr, pw := io.Pipe()
-	err = csvFromItems(pw, items, sp.Schema())
-	if err != nil {
-		return ItemsAsCSV400Response{}, err
-	}
-
 	return ItemsAsCSV200TextcsvResponse{
 		Body: pr,
 	}, nil

server/internal/usecase/interactor/item.go

@@ -4,13 +4,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"time"
 
 	"github.com/reearth/reearth-cms/server/internal/usecase"
 	"github.com/reearth/reearth-cms/server/internal/usecase/gateway"
 	"github.com/reearth/reearth-cms/server/internal/usecase/interfaces"
 	"github.com/reearth/reearth-cms/server/internal/usecase/repo"
 	"github.com/reearth/reearth-cms/server/pkg/event"
+	"github.com/reearth/reearth-cms/server/pkg/group"
 	"github.com/reearth/reearth-cms/server/pkg/id"
 	"github.com/reearth/reearth-cms/server/pkg/item"
 	"github.com/reearth/reearth-cms/server/pkg/request"
@@ -25,6 +27,9 @@ import (
 	"golang.org/x/exp/slices"
 )
 
+const maxPerPage = 100
+const defaultPerPage int64 = 50
+
 type Item struct {
 	repos       *repo.Container
 	gateways    *gateway.Container
@@ -1179,3 +1184,77 @@ func (i Item) getReferencedItems(ctx context.Context, fields []*item.Field) ([]i
 	}
 	return i.repos.Item.FindByIDs(ctx, ids, nil)
 }
+
+func (i Item) ItemsAsCSV(ctx context.Context, modelID id.ModelID, page *int, perPage *int, operator *usecase.Operator) (*io.PipeReader, error) {
+	return Run1(ctx, operator, i.repos, Usecase().Transaction(), func(ctx context.Context) (*io.PipeReader, error) {
+		model, err := i.repos.Model.FindByID(ctx, modelID)
+		if err != nil {
+			return nil, err
+		}
+
+		schemaIDs := id.SchemaIDList{model.Schema()}
+		if model.Metadata() != nil {
+			schemaIDs = append(schemaIDs, *model.Metadata())
+		}
+		schemaList, err := i.repos.Schema.FindByIDs(ctx, schemaIDs)
+		if err != nil {
+			return nil, err
+		}
+		s := schemaList.Schema(lo.ToPtr(model.Schema()))
+		if s == nil {
+			return nil, nil
+		}
+
+		groups, err := i.repos.Group.FindByIDs(ctx, s.Groups())
+		if err != nil {
+			return nil, err
+		}
+
+		groupSchemaList, err := i.repos.Schema.FindByIDs(ctx, groups.SchemaIDs().Add(s.ReferencedSchemas()...))
+		if err != nil {
+			return nil, err
+		}
+		groupSchemaMap := lo.SliceToMap(groups, func(g *group.Group) (id.GroupID, *schema.Schema) {
+			return g.ID(), schemaList.Schema(lo.ToPtr(g.Schema()))
+		})
+		referencedSchemaMap := lo.Map(s.ReferencedSchemas(), func(s schema.ID, _ int) *schema.Schema {
+			return groupSchemaList.Schema(&s)
+		})
+
+		schemaPackage := schema.NewPackage(s, schemaList.Schema(model.Metadata()), groupSchemaMap, referencedSchemaMap)
+
+		// fromPagination
+		p := int64(1)
+		if page != nil && *page > 0 {
+			p = int64(*page)
+		}
+
+		pp := defaultPerPage
+		if perPage != nil {
+			if ppr := *perPage; 1 <= ppr {
+				if ppr > maxPerPage {
+					pp = int64(maxPerPage)
+				} else {
+					pp = int64(ppr)
+				}
+			}
+		}
+
+		paginationOffset := usecasex.OffsetPagination{
+			Offset: (p - 1) * pp,
+			Limit:  pp,
+		}.Wrap()
+
+		items, _, err := i.repos.Item.FindBySchema(ctx, schemaPackage.Schema().ID(), nil, nil, paginationOffset)
+		if err != nil {
+			return nil, err
+		}
+
+		pr, pw := io.Pipe()
+		err = csvFromItems(pw, items, schemaPackage.Schema())
+		if err != nil {
+			return nil, err
+		}
+		return pr, nil
+	})
+}

server/internal/usecase/interactor/item_export.go

@@ -0,0 +1,64 @@
+package interactor
+
+import (
+	"encoding/csv"
+	"io"
+
+	"github.com/labstack/gommon/log"
+	"github.com/reearth/reearth-cms/server/pkg/integrationapi"
+	"github.com/reearth/reearth-cms/server/pkg/item"
+	"github.com/reearth/reearth-cms/server/pkg/schema"
+	"github.com/reearth/reearthx/i18n"
+	"github.com/reearth/reearthx/rerror"
+	"github.com/samber/lo"
+)
+
+var (
+	pointFieldIsNotSupportedError = rerror.NewE(i18n.T("point type is not supported in any geometry field in this model"))
+)
+
+// CSV
+func csvFromItems(pw *io.PipeWriter, l item.VersionedList, s *schema.Schema) error {
+	if !s.IsPointFieldSupported() {
+		return pointFieldIsNotSupportedError
+	}
+
+	go handleCSVGeneration(pw, l, s)
+
+	return nil
+}
+
+func handleCSVGeneration(pw *io.PipeWriter, l item.VersionedList, s *schema.Schema) {
+	err := generateCSV(pw, l, s)
+	if err != nil {
+		log.Errorf("failed to generate CSV: %v", err)
+		_ = pw.CloseWithError(err)
+	} else {
+		_ = pw.Close()
+	}
+}
+
+func generateCSV(pw *io.PipeWriter, l item.VersionedList, s *schema.Schema) error {
+	w := csv.NewWriter(pw)
+	defer w.Flush()
+
+	headers := integrationapi.BuildCSVHeaders(s)
+	if err := w.Write(headers); err != nil {
+		return err
+	}
+
+	nonGeoFields := lo.Filter(s.Fields(), func(f *schema.Field, _ int) bool {
+		return !f.IsGeometryField()
+	})
+
+	for _, ver := range l {
+		row, ok := integrationapi.RowFromItem(ver.Value(), nonGeoFields)
+		if ok {
+			if err := w.Write(row); err != nil {
+				return err
+			}
+		}
+	}
+
+	return w.Error()
+}

server/internal/usecase/interactor/item_export_test.go

@@ -0,0 +1,100 @@
+package interactor
+
+import (
+	"io"
+	"testing"
+
+	"github.com/reearth/reearth-cms/server/pkg/id"
+	"github.com/reearth/reearth-cms/server/pkg/item"
+	"github.com/reearth/reearth-cms/server/pkg/schema"
+	"github.com/reearth/reearth-cms/server/pkg/value"
+	"github.com/reearth/reearth-cms/server/pkg/version"
+	"github.com/reearth/reearthx/account/accountdomain"
+	"github.com/reearth/reearthx/util"
+	"github.com/samber/lo"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCSVFromItems(t *testing.T) {
+	iid := id.NewItemID()
+	sid := id.NewSchemaID()
+	mid := id.NewModelID()
+	tid := id.NewThreadID()
+	pid := id.NewProjectID()
+	gst := schema.GeometryObjectSupportedTypeList{schema.GeometryObjectSupportedTypePoint, schema.GeometryObjectSupportedTypeLineString}
+	gest := schema.GeometryEditorSupportedTypeList{schema.GeometryEditorSupportedTypePoint, schema.GeometryEditorSupportedTypeLineString}
+	sf1 := schema.NewField(schema.NewGeometryObject(gst).TypeProperty()).NewID().Name("geo1").Key(id.RandomKey()).MustBuild()
+	sf3 := schema.NewField(schema.NewGeometryEditor(gest).TypeProperty()).NewID().Name("geo2").Key(id.RandomKey()).MustBuild()
+	in4, _ := schema.NewInteger(lo.ToPtr(int64(1)), lo.ToPtr(int64(100)))
+	tp4 := in4.TypeProperty()
+	sf4 := schema.NewField(tp4).NewID().Name("age").Key(id.RandomKey()).MustBuild()
+	sf5 := schema.NewField(schema.NewBool().TypeProperty()).NewID().Name("isMarried").Key(id.RandomKey()).MustBuild()
+	s1 := schema.New().ID(sid).Fields([]*schema.Field{sf1, sf3, sf4, sf5}).Workspace(accountdomain.NewWorkspaceID()).Project(pid).MustBuild()
+	fi1 := item.NewField(sf1.ID(), value.TypeGeometryObject.Value("{\"coordinates\":[139.28179282584915,36.58570985749664],\"type\":\"Point\"}").AsMultiple(), nil)
+	fi2 := item.NewField(sf3.ID(), value.TypeGeometryEditor.Value("{\"coordinates\":[139.28179282584915,36.58570985749664],\"type\":\"Point\"}").AsMultiple(), nil)
+	fi3 := item.NewField(sf4.ID(), value.TypeInteger.Value(30).AsMultiple(), nil)
+	fi4 := item.NewField(sf5.ID(), value.TypeBool.Value(true).AsMultiple(), nil)
+	i1 := item.New().
+		ID(iid).
+		Schema(sid).
+		Project(pid).
+		Fields([]*item.Field{fi1, fi2, fi3, fi4}).
+		Model(mid).
+		Thread(tid).
+		MustBuild()
+	v1 := version.New()
+	vi1 := version.MustBeValue(v1, nil, version.NewRefs(version.Latest), util.Now(), i1)
+
+	// with geometry fields
+	ver1 := item.VersionedList{vi1}
+	_, pw := io.Pipe()
+	err := csvFromItems(pw, ver1, s1)
+	assert.Nil(t, err)
+
+	// no geometry fields
+	iid2 := id.NewItemID()
+	sid2 := id.NewSchemaID()
+	mid2 := id.NewModelID()
+	tid2 := id.NewThreadID()
+	sf2 := schema.NewField(schema.NewText(lo.ToPtr(10)).TypeProperty()).NewID().Key(id.RandomKey()).MustBuild()
+	s2 := schema.New().ID(sid).Fields([]*schema.Field{sf2}).Workspace(accountdomain.NewWorkspaceID()).Project(pid).MustBuild()
+	i2 := item.New().
+		ID(iid2).
+		Schema(sid2).
+		Project(pid).
+		Fields([]*item.Field{item.NewField(sf2.ID(), value.TypeText.Value("test").AsMultiple(), nil)}).
+		Model(mid2).
+		Thread(tid2).
+		MustBuild()
+	v2 := version.New()
+	vi2 := version.MustBeValue(v2, nil, version.NewRefs(version.Latest), util.Now(), i2)
+	ver2 := item.VersionedList{vi2}
+	expectErr2 := pointFieldIsNotSupportedError
+	_, pw1 := io.Pipe()
+	err = csvFromItems(pw1, ver2, s2)
+	assert.Equal(t, expectErr2, err)
+
+	// point field is not supported
+	iid3 := id.NewItemID()
+	sid3 := id.NewSchemaID()
+	mid3 := id.NewModelID()
+	tid3 := id.NewThreadID()
+	gst2 := schema.GeometryObjectSupportedTypeList{schema.GeometryObjectSupportedTypeLineString, schema.GeometryObjectSupportedTypePolygon}
+	sf6 := schema.NewField(schema.NewGeometryObject(gst2).TypeProperty()).NewID().Name("geo3").Key(id.RandomKey()).MustBuild()
+	s3 := schema.New().ID(sid).Fields([]*schema.Field{sf6}).Workspace(accountdomain.NewWorkspaceID()).Project(pid).MustBuild()
+	i3 := item.New().
+		ID(iid3).
+		Schema(sid3).
+		Project(pid).
+		Fields([]*item.Field{item.NewField(sf6.ID(), value.TypeText.Value("{\n  \"coordinates\": [\n    [\n      139.65439725962517,\n      36.34793305387103\n    ],\n    [\n      139.61688622815393,\n      35.910803456352724\n    ]\n  ],\n  \"type\": \"LineString\"\n}").AsMultiple(), nil)}).
+		Model(mid3).
+		Thread(tid3).
+		MustBuild()
+	v3 := version.New()
+	vi3 := version.MustBeValue(v3, nil, version.NewRefs(version.Latest), util.Now(), i3)
+	ver3 := item.VersionedList{vi3}
+	expectErr3 := pointFieldIsNotSupportedError
+	_, pw2 := io.Pipe()
+	err = csvFromItems(pw2, ver3, s3)
+	assert.Equal(t, expectErr3, err)
+}

server/internal/usecase/interfaces/item.go

@@ -2,6 +2,7 @@ package interfaces
 
 import (
 	"context"
+	"io"
 	"time"
 
 	"github.com/reearth/reearth-cms/server/internal/usecase"
@@ -91,6 +92,7 @@ type Item interface {
 	FindAllVersionsByID(context.Context, id.ItemID, *usecase.Operator) (item.VersionedList, error)
 	Search(context.Context, schema.Package, *item.Query, *usecasex.Pagination, *usecase.Operator) (item.VersionedList, *usecasex.PageInfo, error)
 	ItemStatus(context.Context, id.ItemIDList, *usecase.Operator) (map[id.ItemID]item.Status, error)
+	ItemsAsCSV(context.Context, id.ModelID, *int, *int, *usecase.Operator) (*io.PipeReader, error)
 	LastModifiedByModel(context.Context, id.ModelID, *usecase.Operator) (time.Time, error)
 	IsItemReferenced(context.Context, id.ItemID, id.FieldID, *usecase.Operator) (bool, error)
 	Create(context.Context, CreateItemParam, *usecase.Operator) (item.Versioned, error)