Async Postprocessing

.golangci.yaml

@@ -1,4 +1,6 @@
 issues:
+  exclude:
+    - "Error return value of .* is not checked"
   exclude-rules:
     - path: internal/http/interceptors/log/log.go
       text: "SA1019:"

changelog/unreleased/async-postprocessing.md

@@ -0,0 +1,5 @@
+Enhancement: Async Postprocessing
+
+Provides functionality for async postprocessing. This will allow the system to do the postprocessing (virusscan, copying of bytes to their final destination, ...) asynchronous to the users request. Major change when active.
+
+https://github.com/cs3org/reva/pull/3531

internal/http/services/appprovider/appprovider.go

@@ -384,6 +384,17 @@ func (s *svc) handleOpen(openMode int) http.HandlerFunc {
 			Path:       ".",
 		}
 
+		statRes, err := client.Stat(ctx, &provider.StatRequest{Ref: fileRef})
+		if err != nil {
+			writeError(w, r, appErrorServerError, "Internal error accessing the file, please try again later", err)
+			return
+		}
+
+		if status := utils.ReadPlainFromOpaque(statRes.GetInfo().GetOpaque(), "status"); status == "processing" {
+			writeError(w, r, appErrorTooEarly, "The requested file is not yet available, please try again later", nil)
+			return
+		}
+
 		viewMode, err := getViewModeFromPublicScope(ctx)
 		if err != nil {
 			writeError(w, r, appErrorPermissionDenied, "permission denied to open the application", err)

internal/http/services/appprovider/errors.go

@@ -36,6 +36,7 @@ const (
 	appErrorUnimplemented    appErrorCode = "NOT_IMPLEMENTED"
 	appErrorInvalidParameter appErrorCode = "INVALID_PARAMETER"
 	appErrorServerError      appErrorCode = "SERVER_ERROR"
+	appErrorTooEarly         appErrorCode = "TOO_EARLY"
 )
 
 // appErrorCodeMapping stores the HTTP error code mapping for various APIErrorCodes
@@ -47,6 +48,7 @@ var appErrorCodeMapping = map[appErrorCode]int{
 	appErrorInvalidParameter: http.StatusBadRequest,
 	appErrorServerError:      http.StatusInternalServerError,
 	appErrorPermissionDenied: http.StatusForbidden,
+	appErrorTooEarly:         http.StatusTooEarly,
 }
 
 // APIError encompasses the error type and message

internal/http/services/owncloud/ocdav/propfind/propfind.go

@@ -1476,6 +1476,14 @@ func mdToPropResponse(ctx context.Context, pf *XML, md *provider.ResourceInfo, p
 		}
 	}
 
+	if status := utils.ReadPlainFromOpaque(md.Opaque, "status"); status == "processing" {
+		response.Propstat = append(response.Propstat, PropstatXML{
+			Status: "HTTP/1.1 425 TOO EARLY",
+			Prop:   propstatOK.Prop,
+		})
+		return &response, nil
+	}
+
 	if len(propstatOK.Prop) > 0 {
 		response.Propstat = append(response.Propstat, propstatOK)
 	}

internal/http/services/owncloud/ocdav/put.go

@@ -137,6 +137,26 @@ func (s *svc) handlePut(ctx context.Context, w http.ResponseWriter, r *http.Requ
 		return
 	}
 
+	/* FIXME: to bring back 0-byte touch instead upload return fileid in TouchFileRequest and add it to response headers
+	if length == 0 {
+		tfRes, err := s.gwClient.TouchFile(ctx, &provider.TouchFileRequest{
+			Ref: ref,
+		})
+		if err != nil {
+			log.Error().Err(err).Msg("error sending grpc touch file request")
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+		if tfRes.Status.Code != rpc.Code_CODE_OK {
+			log.Error().Interface("status", tfRes.Status).Msg("error touching file")
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+		w.WriteHeader(http.StatusCreated)
+		return
+	}
+	*/
+
 	opaqueMap := map[string]*typespb.OpaqueEntry{
 		net.HeaderUploadLength: {
 			Decoder: "plain",

pkg/events/postprocessing.go

@@ -0,0 +1,164 @@
+// Copyright 2018-2022 CERN
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// In applying this license, CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+package events
+
+import (
+	"encoding/json"
+	"time"
+
+	user "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+)
+
+type (
+	// Postprocessingstep are the available postprocessingsteps
+	Postprocessingstep string
+
+	// PostprocessingOutcome defines the result of the postprocessing
+	PostprocessingOutcome string
+)
+
+var (
+	// PPStepAntivirus is the step that scans for viruses
+	PPStepAntivirus Postprocessingstep = "virusscan"
+	// PPStepFTS is the step that indexes files for full text search
+	PPStepFTS Postprocessingstep = "fts"
+	// PPStepDelay is the step that processing. Useful for testing or user annoyment
+	PPStepDelay Postprocessingstep = "delay"
+
+	// PPOutcomeDelete means that the file and the upload should be deleted
+	PPOutcomeDelete PostprocessingOutcome = "delete"
+	// PPOutcomeAbort means that the upload is cancelled but the bytes are being kept in the upload folder
+	PPOutcomeAbort PostprocessingOutcome = "abort"
+	// PPOutcomeContinue means that the upload is moved to its final destination (eventually being marked with pp results)
+	PPOutcomeContinue PostprocessingOutcome = "continue"
+)
+
+// BytesReceived is emitted by the server when it received all bytes of an upload
+type BytesReceived struct {
+	UploadID      string
+	SpaceOwner    *user.UserId
+	ExecutingUser *user.User
+	ResourceID    *provider.ResourceId
+	Filename      string
+	Filesize      uint64
+	URL           string
+}
+
+// Unmarshal to fulfill umarshaller interface
+func (BytesReceived) Unmarshal(v []byte) (interface{}, error) {
+	e := BytesReceived{}
+	err := json.Unmarshal(v, &e)
+	return e, err
+}
+
+// VirusscanFinished is emitted by the server when it has completed an antivirus scan
+type VirusscanFinished struct {
+	Infected      bool
+	Outcome       PostprocessingOutcome
+	UploadID      string
+	Filename      string
+	ExecutingUser *user.User
+	Description   string
+	Scandate      time.Time
+	ResourceID    *provider.ResourceId
+	ErrorMsg      string // empty when no error
+}
+
+// Unmarshal to fulfill umarshaller interface
+func (VirusscanFinished) Unmarshal(v []byte) (interface{}, error) {
+	e := VirusscanFinished{}
+	err := json.Unmarshal(v, &e)
+	return e, err
+}
+
+// StartPostprocessingStep can be issued by the server to start a postprocessing step
+type StartPostprocessingStep struct {
+	UploadID      string
+	URL           string
+	ExecutingUser *user.User
+	Filename      string
+	Filesize      uint64
+	Token         string               // for file retrieval in after upload case
+	ResourceID    *provider.ResourceId // for file retrieval in after upload case
+	RevaToken     string               // for file retrieval in after upload case
+
+	StepToStart Postprocessingstep
+}
+
+// Unmarshal to fulfill umarshaller interface
+func (StartPostprocessingStep) Unmarshal(v []byte) (interface{}, error) {
+	e := StartPostprocessingStep{}
+	err := json.Unmarshal(v, &e)
+	return e, err
+}
+
+// PostprocessingStepFinished can be issued by the server when a postprocessing step is finished
+type PostprocessingStepFinished struct {
+	UploadID      string
+	ExecutingUser *user.User
+	Filename      string
+
+	FinishedStep Postprocessingstep    // name of the step
+	Result       interface{}           // result information
+	Error        error                 // possible error of the step
+	Outcome      PostprocessingOutcome // some services may cause postprocessing to stop
+}
+
+// Unmarshal to fulfill umarshaller interface
+func (PostprocessingStepFinished) Unmarshal(v []byte) (interface{}, error) {
+	e := PostprocessingStepFinished{}
+	err := json.Unmarshal(v, &e)
+	return e, err
+}
+
+// PostprocessingFinished is emitted by *some* service which can decide that
+type PostprocessingFinished struct {
+	UploadID      string
+	Filename      string
+	SpaceOwner    *user.UserId
+	ExecutingUser *user.User
+	Result        map[Postprocessingstep]interface{} // it is a map[step]Event
+	Outcome       PostprocessingOutcome
+}
+
+// Unmarshal to fulfill umarshaller interface
+func (PostprocessingFinished) Unmarshal(v []byte) (interface{}, error) {
+	e := PostprocessingFinished{}
+	err := json.Unmarshal(v, &e)
+	return e, err
+}
+
+// UploadReady is emitted by the storage provider when postprocessing is finished
+type UploadReady struct {
+	UploadID      string
+	Filename      string
+	SpaceOwner    *user.UserId
+	ExecutingUser *user.User
+	FileRef       *provider.Reference
+	Failed        bool
+	// add reference here? We could use it to inform client pp is finished
+}
+
+// Unmarshal to fulfill umarshaller interface
+func (UploadReady) Unmarshal(v []byte) (interface{}, error) {
+	e := UploadReady{}
+	err := json.Unmarshal(v, &e)
+	return e, err
+}