Space Trash-bin expiration cli (#5500)

* add storage-users trash-bin cli
add task to clean up outdated trash-bin resources
add trash-bin cli purge-expired command to purge expired trash-bin resources
add purge-expired task tests

services/storage-users/pkg/event/event.go

@@ -0,0 +1,51 @@
+package event
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"os"
+
+	"github.com/cs3org/reva/v2/pkg/events/stream"
+	"github.com/go-micro/plugins/v4/events/natsjs"
+	ociscrypto "github.com/owncloud/ocis/v2/ocis-pkg/crypto"
+	"github.com/owncloud/ocis/v2/services/storage-users/pkg/config"
+	"go-micro.dev/v4/events"
+)
+
+// NewStream prepares the requested nats stream and returns it.
+func NewStream(cfg config.Events) (events.Stream, error) {
+	var tlsConf *tls.Config
+
+	if cfg.EnableTLS {
+		var rootCAPool *x509.CertPool
+		if cfg.TLSRootCaCertPath != "" {
+			rootCrtFile, err := os.Open(cfg.TLSRootCaCertPath)
+			if err != nil {
+				return nil, err
+			}
+
+			rootCAPool, err = ociscrypto.NewCertPoolFromPEM(rootCrtFile)
+			if err != nil {
+				return nil, err
+			}
+			cfg.TLSInsecure = false
+		}
+
+		tlsConf = &tls.Config{
+			MinVersion: tls.VersionTLS12,
+			RootCAs:    rootCAPool,
+		}
+	}
+
+	s, err := stream.Nats(
+		natsjs.TLSConfig(tlsConf),
+		natsjs.Address(cfg.Addr),
+		natsjs.ClusterID(cfg.ClusterID),
+	)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return s, nil
+}

services/storage-users/pkg/event/service.go

@@ -0,0 +1,85 @@
+package event
+
+import (
+	"time"
+
+	apiGateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
+	apiUser "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+	"github.com/cs3org/reva/v2/pkg/events"
+	"github.com/owncloud/ocis/v2/ocis-pkg/log"
+	"github.com/owncloud/ocis/v2/services/storage-users/pkg/config"
+	"github.com/owncloud/ocis/v2/services/storage-users/pkg/task"
+)
+
+const (
+	consumerGroup = "storage-users"
+)
+
+// Service wraps all common logic that is needed to react to incoming events.
+type Service struct {
+	gatewayClient apiGateway.GatewayAPIClient
+	eventStream   events.Stream
+	logger        log.Logger
+	config        config.Config
+}
+
+// NewService prepares and returns a Service implementation.
+func NewService(gatewayClient apiGateway.GatewayAPIClient, eventStream events.Stream, logger log.Logger, conf config.Config) (Service, error) {
+	svc := Service{
+		gatewayClient: gatewayClient,
+		eventStream:   eventStream,
+		logger:        logger,
+		config:        conf,
+	}
+
+	return svc, nil
+}
+
+// Run to fulfil Runner interface
+func (s Service) Run() error {
+	ch, err := events.Consume(s.eventStream, consumerGroup, PurgeTrashBin{})
+	if err != nil {
+		return err
+	}
+
+	for e := range ch {
+		var errs []error
+
+		switch ev := e.(type) {
+		case PurgeTrashBin:
+			executionTime := ev.ExecutionTime
+			if executionTime.IsZero() {
+				executionTime = time.Now()
+			}
+
+			executantID := ev.ExecutantID
+			if executantID == nil {
+				executantID = &apiUser.UserId{OpaqueId: s.config.Tasks.PurgeTrashBin.UserID}
+			}
+
+			tasks := map[task.SpaceType]time.Time{
+				task.Project:  executionTime.Add(-s.config.Tasks.PurgeTrashBin.ProjectDeleteBefore),
+				task.Personal: executionTime.Add(-s.config.Tasks.PurgeTrashBin.PersonalDeleteBefore),
+			}
+
+			for spaceType, deleteBefore := range tasks {
+				// skip task execution if the deleteBefore time is the same as the now time,
+				// which indicates that the duration configuration for this space type is set to 0 which is the equivalent to disabled.
+				if deleteBefore.Equal(executionTime) {
+					continue
+				}
+
+				if err = task.PurgeTrashBin(executantID, deleteBefore, spaceType, s.gatewayClient, s.config.Commons.MachineAuthAPIKey); err != nil {
+					errs = append(errs, err)
+				}
+			}
+
+		}
+
+		for _, err := range errs {
+			s.logger.Error().Err(err).Interface("event", e)
+		}
+	}
+
+	return nil
+}

services/storage-users/pkg/event/trigger.go

@@ -0,0 +1,21 @@
+package event
+
+import (
+	"encoding/json"
+	"time"
+
+	apiUser "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+)
+
+// PurgeTrashBin wraps all needed information to purge a trash-bin
+type PurgeTrashBin struct {
+	ExecutantID   *apiUser.UserId
+	ExecutionTime time.Time
+}
+
+// Unmarshal to fulfill umarshaller interface
+func (PurgeTrashBin) Unmarshal(v []byte) (interface{}, error) {
+	e := PurgeTrashBin{}
+	err := json.Unmarshal(v, &e)
+	return e, err
+}