feat: [OCISDEV-237] the sse support added for the public links

Author: 2403905

changelog/unreleased/enhc-sse-support.md

@@ -0,0 +1,5 @@
+Enhancement: The sse support added for the public links
+
+The sse support added for the public links
+
+https://github.com/owncloud/ocis/pull/11623

services/clientlog/pkg/service/service.go

@@ -13,10 +13,12 @@ import (
 	group "github.com/cs3org/go-cs3apis/cs3/identity/group/v1beta1"
 	user "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
 	rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
+	link "github.com/cs3org/go-cs3apis/cs3/sharing/link/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"go.opentelemetry.io/otel/trace"
 
 	"github.com/owncloud/reva/v2/pkg/events"
+	"github.com/owncloud/reva/v2/pkg/publicshare"
 	"github.com/owncloud/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/owncloud/reva/v2/pkg/storagespace"
 	"github.com/owncloud/reva/v2/pkg/utils"
@@ -132,6 +134,8 @@ func (cl *ClientlogService) processEvent(event events.Event) {
 		users, data, err = processShareEvent(ctx, ref, gwc, event.InitiatorID, uid, gid)
 	}
 
+	tokens := make([]string, 0)
+
 	switch e := event.Event.(type) {
 	default:
 		err = errors.New("unhandled event")
@@ -141,26 +145,34 @@ func (cl *ClientlogService) processEvent(event events.Event) {
 			return
 		}
 		fileEv("postprocessing-finished", e.FileRef)
+		tokens, _ = listPublicShareTokens(ctx, gwc, e.FileRef, cl.log)
 	case events.ItemTrashed:
 		evType = "item-trashed"
 		users, data, err = processItemTrashedEvent(ctx, e.Ref, gwc, event.InitiatorID, e.ID)
+		tokens, _ = listPublicShareTokens(ctx, gwc, e.Ref, cl.log)
 	case events.ItemRestored:
 		fileEv("item-restored", e.Ref)
+		tokens, _ = listPublicShareTokens(ctx, gwc, e.Ref, cl.log)
 	case events.ContainerCreated:
 		fileEv("folder-created", e.Ref)
+		tokens, _ = listPublicShareTokens(ctx, gwc, e.Ref, cl.log)
 	case events.ItemMoved:
 		// we send a dedicated event in case the item was only renamed
 		if isRename(e.OldReference, e.Ref) {
 			fileEv("item-renamed", e.Ref)
 		} else {
 			fileEv("item-moved", e.Ref)
 		}
+		tokens, _ = listPublicShareTokens(ctx, gwc, e.Ref, cl.log)
 	case events.FileLocked:
 		fileEv("file-locked", e.Ref)
+		tokens, _ = listPublicShareTokens(ctx, gwc, e.Ref, cl.log)
 	case events.FileUnlocked:
 		fileEv("file-unlocked", e.Ref)
+		tokens, _ = listPublicShareTokens(ctx, gwc, e.Ref, cl.log)
 	case events.FileTouched:
 		fileEv("file-touched", e.Ref)
+		tokens, _ = listPublicShareTokens(ctx, gwc, e.Ref, cl.log)
 	case events.SpaceShared:
 		r, _ := storagespace.ParseReference(e.ID.GetOpaqueId())
 		shareEv("space-member-added", &r, e.GranteeUserID, e.GranteeGroupID)
@@ -192,6 +204,9 @@ func (cl *ClientlogService) processEvent(event events.Event) {
 	}
 
 	// II) instruct sse service to send the information
+	if len(tokens) > 0 {
+		users = append(users, tokens...)
+	}
 	if err := cl.sendSSE(users, evType, data); err != nil {
 		cl.log.Error().Err(err).Interface("userIDs", users).Str("eventid", event.ID).Msg("failed to store event for user")
 		return
@@ -320,3 +335,41 @@ func backchannelLogoutEvent(e events.BackchannelLogout) (string, []string, Backc
 		Timestamp: e.Timestamp.String(),
 	}
 }
+
+func listPublicShareTokens(ctx context.Context, gwc gateway.GatewayAPIClient, ref *provider.Reference, log log.Logger) ([]string, error) {
+	tokens := make([]string, 0)
+	info, err := utils.GetResource(ctx, ref, gwc)
+	if err != nil {
+		log.Error().Err(err).Str("event", "listPublicShareTokens").Msg("error getting ResourceInfo")
+		return tokens, err
+	}
+
+	if utils.IsSpaceRoot(info) {
+		return tokens, nil
+	}
+
+	req := link.ListPublicSharesRequest{
+		Filters: []*link.ListPublicSharesRequest_Filter{publicshare.ResourceIDFilter(info.GetId())},
+	}
+
+	res, err := gwc.ListPublicShares(ctx, &req)
+	if err != nil {
+		log.Error().Err(err).Str("event", "listPublicShareTokens").Msg("error getting ListPublicShares")
+		return tokens, err
+	}
+
+	if res.GetStatus().GetCode() != rpc.Code_CODE_OK {
+		log.Error().Str("event", "listPublicShareTokens").Msg("error getting ListPublicShares " + res.GetStatus().GetCode().String())
+		return tokens, fmt.Errorf("error getting ListPublicShares %s", res.GetStatus().GetCode().String())
+	}
+	for _, s := range res.GetShare() {
+		tokens = append(tokens, s.GetToken())
+	}
+
+	t, err := listPublicShareTokens(ctx, gwc, &provider.Reference{ResourceId: info.GetParentId()}, log)
+	if err != nil {
+		log.Error().Err(err).Str("event", "listPublicShareTokens").Msg("error getting parent listPublicShareTokens")
+		return tokens, err
+	}
+	return append(tokens, t...), nil
+}

services/sse/pkg/service/service.go

@@ -4,11 +4,14 @@ import (
 	"net/http"
 	"time"
 
+	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+
 	"github.com/go-chi/chi/v5"
 	"github.com/r3labs/sse/v2"
 
 	revactx "github.com/owncloud/reva/v2/pkg/ctx"
 	"github.com/owncloud/reva/v2/pkg/events"
+	"github.com/owncloud/reva/v2/pkg/utils"
 
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
 	"github.com/owncloud/ocis/v2/services/sse/pkg/config"
@@ -73,6 +76,17 @@ func (s SSE) HandleSSE(w http.ResponseWriter, r *http.Request) {
 	}
 
 	uid := u.GetId().GetOpaqueId()
+
+	if len(utils.ReadPlainFromOpaque(u.Opaque, "public-share-role")) > 0 {
+		var impersonator userpb.User
+		if err := utils.ReadJSONFromOpaque(u.Opaque, "impersonating-user", &impersonator); err != nil {
+			s.l.Error().Msg("sse: public share token not found")
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+		uid = impersonator.GetId().GetOpaqueId()
+	}
+
 	if uid == "" {
 		s.l.Error().Msg("sse: user in context is broken")
 		w.WriteHeader(http.StatusInternalServerError)