gtsocial-umbx

federating_wrapped_callbacks.go (28036B)

---

 package pub
 
 import (
 	"context"
 	"encoding/json"
 	"fmt"
 	"net/url"
 
 	"github.com/superseriousbusiness/activity/streams"
 	"github.com/superseriousbusiness/activity/streams/vocab"
 )
 
 // OnFollowBehavior enumerates the different default actions that the go-fed
 // library can provide when receiving a Follow Activity from a peer.
 type OnFollowBehavior int
 
 const (
 	// OnFollowDoNothing does not take any action when a Follow Activity
 	// is received.
 	OnFollowDoNothing OnFollowBehavior = iota
 	// OnFollowAutomaticallyAccept triggers the side effect of sending an
 	// Accept of this Follow request in response.
 	OnFollowAutomaticallyAccept
 	// OnFollowAutomaticallyAccept triggers the side effect of sending a
 	// Reject of this Follow request in response.
 	OnFollowAutomaticallyReject
 )
 
 // FederatingWrappedCallbacks lists the callback functions that already have
 // some side effect behavior provided by the pub library.
 //
 // These functions are wrapped for the Federating Protocol.
 type FederatingWrappedCallbacks struct {
 	// Create handles additional side effects for the Create ActivityStreams
 	// type, specific to the application using go-fed.
 	//
 	// The wrapping callback for the Federating Protocol ensures the
 	// 'object' property is created in the database.
 	//
 	// Create calls Create for each object in the federated Activity.
 	Create func(context.Context, vocab.ActivityStreamsCreate) error
 	// Update handles additional side effects for the Update ActivityStreams
 	// type, specific to the application using go-fed.
 	//
 	// The wrapping callback for the Federating Protocol ensures the
 	// 'object' property is updated in the database.
 	//
 	// Update calls Update on the federated entry from the database, with a
 	// new value.
 	Update func(context.Context, vocab.ActivityStreamsUpdate) error
 	// Delete handles additional side effects for the Delete ActivityStreams
 	// type, specific to the application using go-fed.
 	//
 	// Delete removes the federated entry from the database.
 	Delete func(context.Context, vocab.ActivityStreamsDelete) error
 	// Follow handles additional side effects for the Follow ActivityStreams
 	// type, specific to the application using go-fed.
 	//
 	// The wrapping function can have one of several default behaviors,
 	// depending on the value of the OnFollow setting.
 	Follow func(context.Context, vocab.ActivityStreamsFollow) error
 	// OnFollow determines what action to take for this particular callback
 	// if a Follow Activity is handled.
 	OnFollow OnFollowBehavior
 	// Accept handles additional side effects for the Accept ActivityStreams
 	// type, specific to the application using go-fed.
 	//
 	// The wrapping function determines if this 'Accept' is in response to a
 	// 'Follow'. If so, then the 'actor' is added to the original 'actor's
 	// 'following' collection.
 	//
 	// Otherwise, no side effects are done by go-fed.
 	Accept func(context.Context, vocab.ActivityStreamsAccept) error
 	// Reject handles additional side effects for the Reject ActivityStreams
 	// type, specific to the application using go-fed.
 	//
 	// The wrapping function has no default side effects. However, if this
 	// 'Reject' is in response to a 'Follow' then the client MUST NOT go
 	// forward with adding the 'actor' to the original 'actor's 'following'
 	// collection by the client application.
 	Reject func(context.Context, vocab.ActivityStreamsReject) error
 	// Add handles additional side effects for the Add ActivityStreams
 	// type, specific to the application using go-fed.
 	//
 	// The wrapping function will add the 'object' IRIs to a specific
 	// 'target' collection if the 'target' collection(s) live on this
 	// server.
 	Add func(context.Context, vocab.ActivityStreamsAdd) error
 	// Remove handles additional side effects for the Remove ActivityStreams
 	// type, specific to the application using go-fed.
 	//
 	// The wrapping function will remove all 'object' IRIs from a specific
 	// 'target' collection if the 'target' collection(s) live on this
 	// server.
 	Remove func(context.Context, vocab.ActivityStreamsRemove) error
 	// Like handles additional side effects for the Like ActivityStreams
 	// type, specific to the application using go-fed.
 	//
 	// The wrapping function will add the activity to the "likes" collection
 	// on all 'object' targets owned by this server.
 	Like func(context.Context, vocab.ActivityStreamsLike) error
 	// Announce handles additional side effects for the Announce
 	// ActivityStreams type, specific to the application using go-fed.
 	//
 	// The wrapping function will add the activity to the "shares"
 	// collection on all 'object' targets owned by this server.
 	Announce func(context.Context, vocab.ActivityStreamsAnnounce) error
 	// Undo handles additional side effects for the Undo ActivityStreams
 	// type, specific to the application using go-fed.
 	//
 	// The wrapping function ensures the 'actor' on the 'Undo'
 	// is be the same as the 'actor' on all Activities being undone.
 	// It enforces that the actors on the Undo must correspond to all of the
 	// 'object' actors in some manner.
 	//
 	// It is expected that the application will implement the proper
 	// reversal of activities that are being undone.
 	Undo func(context.Context, vocab.ActivityStreamsUndo) error
 	// Block handles additional side effects for the Block ActivityStreams
 	// type, specific to the application using go-fed.
 	//
 	// The wrapping function provides no default side effects. It simply
 	// calls the wrapped function. However, note that Blocks should not be
 	// received from a federated peer, as delivering Blocks explicitly
 	// deviates from the original ActivityPub specification.
 	Block func(context.Context, vocab.ActivityStreamsBlock) error
 
 	// Sidechannel data -- this is set at request handling time. These must
 	// be set before the callbacks are used.
 
 	// db is the Database the FederatingWrappedCallbacks should use.
 	db Database
 	// inboxIRI is the inboxIRI that is handling this callback.
 	inboxIRI *url.URL
 	// addNewIds creates new 'id' entries on an activity and its objects if
 	// it is a Create activity.
 	addNewIds func(c context.Context, activity Activity) error
 	// deliver delivers an outgoing message.
 	deliver func(c context.Context, outboxIRI *url.URL, activity Activity) error
 	// newTransport creates a new Transport.
 	newTransport func(c context.Context, actorBoxIRI *url.URL, gofedAgent string) (t Transport, err error)
 }
 
 // callbacks returns the WrappedCallbacks members into a single interface slice
 // for use in streams.Resolver callbacks.
 //
 // If the given functions have a type that collides with the default behavior,
 // then disable our default behavior
 func (w FederatingWrappedCallbacks) callbacks(fns []interface{}) []interface{} {
 	enableCreate := true
 	enableUpdate := true
 	enableDelete := true
 	enableFollow := true
 	enableAccept := true
 	enableReject := true
 	enableAdd := true
 	enableRemove := true
 	enableLike := true
 	enableAnnounce := true
 	enableUndo := true
 	enableBlock := true
 	for _, fn := range fns {
 		switch fn.(type) {
 		default:
 			continue
 		case func(context.Context, vocab.ActivityStreamsCreate) error:
 			enableCreate = false
 		case func(context.Context, vocab.ActivityStreamsUpdate) error:
 			enableUpdate = false
 		case func(context.Context, vocab.ActivityStreamsDelete) error:
 			enableDelete = false
 		case func(context.Context, vocab.ActivityStreamsFollow) error:
 			enableFollow = false
 		case func(context.Context, vocab.ActivityStreamsAccept) error:
 			enableAccept = false
 		case func(context.Context, vocab.ActivityStreamsReject) error:
 			enableReject = false
 		case func(context.Context, vocab.ActivityStreamsAdd) error:
 			enableAdd = false
 		case func(context.Context, vocab.ActivityStreamsRemove) error:
 			enableRemove = false
 		case func(context.Context, vocab.ActivityStreamsLike) error:
 			enableLike = false
 		case func(context.Context, vocab.ActivityStreamsAnnounce) error:
 			enableAnnounce = false
 		case func(context.Context, vocab.ActivityStreamsUndo) error:
 			enableUndo = false
 		case func(context.Context, vocab.ActivityStreamsBlock) error:
 			enableBlock = false
 		}
 	}
 	if enableCreate {
 		fns = append(fns, w.create)
 	}
 	if enableUpdate {
 		fns = append(fns, w.update)
 	}
 	if enableDelete {
 		fns = append(fns, w.deleteFn)
 	}
 	if enableFollow {
 		fns = append(fns, w.follow)
 	}
 	if enableAccept {
 		fns = append(fns, w.accept)
 	}
 	if enableReject {
 		fns = append(fns, w.reject)
 	}
 	if enableAdd {
 		fns = append(fns, w.add)
 	}
 	if enableRemove {
 		fns = append(fns, w.remove)
 	}
 	if enableLike {
 		fns = append(fns, w.like)
 	}
 	if enableAnnounce {
 		fns = append(fns, w.announce)
 	}
 	if enableUndo {
 		fns = append(fns, w.undo)
 	}
 	if enableBlock {
 		fns = append(fns, w.block)
 	}
 	return fns
 }
 
 // create implements the federating Create activity side effects.
 func (w FederatingWrappedCallbacks) create(c context.Context, a vocab.ActivityStreamsCreate) error {
 	op := a.GetActivityStreamsObject()
 	if op == nil || op.Len() == 0 {
 		return ErrObjectRequired
 	}
 	// Create anonymous loop function to be able to properly scope the defer
 	// for the database lock at each iteration.
 	loopFn := func(iter vocab.ActivityStreamsObjectPropertyIterator) error {
 		t := iter.GetType()
 		if t == nil && iter.IsIRI() {
 			// Attempt to dereference the IRI instead
 			tport, err := w.newTransport(c, w.inboxIRI, goFedUserAgent())
 			if err != nil {
 				return err
 			}
 			b, err := tport.Dereference(c, iter.GetIRI())
 			if err != nil {
 				return err
 			}
 			var m map[string]interface{}
 			if err = json.Unmarshal(b, &m); err != nil {
 				return err
 			}
 			t, err = streams.ToType(c, m)
 			if err != nil {
 				return err
 			}
 		} else if t == nil {
 			return fmt.Errorf("cannot handle federated create: object is neither a value nor IRI")
 		}
 		id, err := GetId(t)
 		if err != nil {
 			return err
 		}
 		var unlock func()
 		unlock, err = w.db.Lock(c, id)
 		if err != nil {
 			return err
 		}
 		defer unlock()
 		if err := w.db.Create(c, t); err != nil {
 			return err
 		}
 		return nil
 	}
 	for iter := op.Begin(); iter != op.End(); iter = iter.Next() {
 		if err := loopFn(iter); err != nil {
 			return err
 		}
 	}
 	if w.Create != nil {
 		return w.Create(c, a)
 	}
 	return nil
 }
 
 // update implements the federating Update activity side effects.
 func (w FederatingWrappedCallbacks) update(c context.Context, a vocab.ActivityStreamsUpdate) error {
 	op := a.GetActivityStreamsObject()
 	if op == nil || op.Len() == 0 {
 		return ErrObjectRequired
 	}
 	if err := mustHaveActivityOriginMatchObjects(a); err != nil {
 		return err
 	}
 	// Create anonymous loop function to be able to properly scope the defer
 	// for the database lock at each iteration.
 	loopFn := func(iter vocab.ActivityStreamsObjectPropertyIterator) error {
 		t := iter.GetType()
 		if t == nil {
 			return fmt.Errorf("update requires an object to be wholly provided")
 		}
 		id, err := GetId(t)
 		if err != nil {
 			return err
 		}
 		var unlock func()
 		unlock, err = w.db.Lock(c, id)
 		if err != nil {
 			return err
 		}
 		defer unlock()
 		if err := w.db.Update(c, t); err != nil {
 			return err
 		}
 		return nil
 	}
 	for iter := op.Begin(); iter != op.End(); iter = iter.Next() {
 		if err := loopFn(iter); err != nil {
 			return err
 		}
 	}
 	if w.Update != nil {
 		return w.Update(c, a)
 	}
 	return nil
 }
 
 // deleteFn implements the federating Delete activity side effects.
 func (w FederatingWrappedCallbacks) deleteFn(c context.Context, a vocab.ActivityStreamsDelete) error {
 	op := a.GetActivityStreamsObject()
 	if op == nil || op.Len() == 0 {
 		return ErrObjectRequired
 	}
 	if err := mustHaveActivityOriginMatchObjects(a); err != nil {
 		return err
 	}
 	// Create anonymous loop function to be able to properly scope the defer
 	// for the database lock at each iteration.
 	loopFn := func(iter vocab.ActivityStreamsObjectPropertyIterator) error {
 		id, err := ToId(iter)
 		if err != nil {
 			return err
 		}
 		var unlock func()
 		unlock, err = w.db.Lock(c, id)
 		if err != nil {
 			return err
 		}
 		defer unlock()
 		if err := w.db.Delete(c, id); err != nil {
 			return err
 		}
 		return nil
 	}
 	for iter := op.Begin(); iter != op.End(); iter = iter.Next() {
 		if err := loopFn(iter); err != nil {
 			return err
 		}
 	}
 	if w.Delete != nil {
 		return w.Delete(c, a)
 	}
 	return nil
 }
 
 // follow implements the federating Follow activity side effects.
 func (w FederatingWrappedCallbacks) follow(c context.Context, a vocab.ActivityStreamsFollow) error {
 	op := a.GetActivityStreamsObject()
 	if op == nil || op.Len() == 0 {
 		return ErrObjectRequired
 	}
 	// Check that we own at least one of the 'object' properties, and ensure
 	// it is to the actor that owns this inbox.
 	//
 	// If not then don't send a response. It was federated to us as an FYI,
 	// by mistake, or some other reason.
 	unlock, err := w.db.Lock(c, w.inboxIRI)
 	if err != nil {
 		return err
 	}
 	// WARNING: Unlock not deferred.
 	actorIRI, err := w.db.ActorForInbox(c, w.inboxIRI)
 	unlock() // unlock even on error
 	if err != nil {
 		return err
 	}
 	// Unlock must be called by now and every branch above.
 	isMe := false
 	if w.OnFollow != OnFollowDoNothing {
 		for iter := op.Begin(); iter != op.End(); iter = iter.Next() {
 			id, err := ToId(iter)
 			if err != nil {
 				return err
 			}
 			if id.String() == actorIRI.String() {
 				isMe = true
 				break
 			}
 		}
 	}
 	if isMe {
 		// Prepare the response.
 		var response Activity
 		if w.OnFollow == OnFollowAutomaticallyAccept {
 			response = streams.NewActivityStreamsAccept()
 		} else if w.OnFollow == OnFollowAutomaticallyReject {
 			response = streams.NewActivityStreamsReject()
 		} else {
 			return fmt.Errorf("unknown OnFollowBehavior: %d", w.OnFollow)
 		}
 		// Set us as the 'actor'.
 		me := streams.NewActivityStreamsActorProperty()
 		response.SetActivityStreamsActor(me)
 		me.AppendIRI(actorIRI)
 		// Set the Follow as the 'object' property.
 		op := streams.NewActivityStreamsObjectProperty()
 		response.SetActivityStreamsObject(op)
 		op.AppendActivityStreamsFollow(a)
 		// Add all actors on the original Follow to the 'to' property.
 		recipients := make([]*url.URL, 0)
 		to := streams.NewActivityStreamsToProperty()
 		response.SetActivityStreamsTo(to)
 		followActors := a.GetActivityStreamsActor()
 		for iter := followActors.Begin(); iter != followActors.End(); iter = iter.Next() {
 			id, err := ToId(iter)
 			if err != nil {
 				return err
 			}
 			to.AppendIRI(id)
 			recipients = append(recipients, id)
 		}
 		if w.OnFollow == OnFollowAutomaticallyAccept {
 			// If automatically accepting, then also update our
 			// followers collection with the new actors.
 			//
 			// If automatically rejecting, do not update the
 			// followers collection.
 			unlock, err := w.db.Lock(c, actorIRI)
 			if err != nil {
 				return err
 			}
 			// WARNING: Unlock not deferred.
 			followers, err := w.db.Followers(c, actorIRI)
 			if err != nil {
 				unlock()
 				return err
 			}
 			items := followers.GetActivityStreamsItems()
 			if items == nil {
 				items = streams.NewActivityStreamsItemsProperty()
 				followers.SetActivityStreamsItems(items)
 			}
 			for _, elem := range recipients {
 				items.PrependIRI(elem)
 			}
 			err = w.db.Update(c, followers)
 			unlock() // unlock even on error
 			if err != nil {
 				return err
 			}
 			// Unlock must be called by now and every branch above.
 		}
 		// Lock without defer!
 		unlock, err := w.db.Lock(c, w.inboxIRI)
 		if err != nil {
 			return err
 		}
 		outboxIRI, err := w.db.OutboxForInbox(c, w.inboxIRI)
 		unlock() // unlock after, regardless
 		if err != nil {
 			return err
 		}
 		// Everything must be unlocked by now.
 		if err := w.addNewIds(c, response); err != nil {
 			return err
 		} else if err := w.deliver(c, outboxIRI, response); err != nil {
 			return err
 		}
 	}
 	if w.Follow != nil {
 		return w.Follow(c, a)
 	}
 	return nil
 }
 
 // accept implements the federating Accept activity side effects.
 func (w FederatingWrappedCallbacks) accept(c context.Context, a vocab.ActivityStreamsAccept) error {
 	op := a.GetActivityStreamsObject()
 	if op != nil && op.Len() > 0 {
 		// Get this actor's id.
 		unlock, err := w.db.Lock(c, w.inboxIRI)
 		if err != nil {
 			return err
 		}
 		// WARNING: Unlock not deferred.
 		actorIRI, err := w.db.ActorForInbox(c, w.inboxIRI)
 		unlock() // unlock after regardless
 		if err != nil {
 			return err
 		}
 		// Unlock must be called by now and every branch above.
 		//
 		// Determine if we are in a follow on the 'object' property.
 		//
 		// TODO: Handle Accept multiple Follow.
 		var maybeMyFollowIRI *url.URL
 		for iter := op.Begin(); iter != op.End(); iter = iter.Next() {
 			t := iter.GetType()
 			if t == nil && iter.IsIRI() {
 				// Attempt to dereference the IRI instead
 				tport, err := w.newTransport(c, w.inboxIRI, goFedUserAgent())
 				if err != nil {
 					return err
 				}
 				b, err := tport.Dereference(c, iter.GetIRI())
 				if err != nil {
 					return err
 				}
 				var m map[string]interface{}
 				if err = json.Unmarshal(b, &m); err != nil {
 					return err
 				}
 				t, err = streams.ToType(c, m)
 				if err != nil {
 					return err
 				}
 			} else if t == nil {
 				return fmt.Errorf("cannot handle federated create: object is neither a value nor IRI")
 			}
 			// Ensure it is a Follow.
 			if !streams.IsOrExtendsActivityStreamsFollow(t) {
 				continue
 			}
 			follow, ok := t.(Activity)
 			if !ok {
 				return fmt.Errorf("a Follow in an Accept does not satisfy the Activity interface")
 			}
 			followId, err := GetId(follow)
 			if err != nil {
 				return err
 			}
 			// Ensure that we are one of the actors on the Follow.
 			actors := follow.GetActivityStreamsActor()
 			for iter := actors.Begin(); iter != actors.End(); iter = iter.Next() {
 				id, err := ToId(iter)
 				if err != nil {
 					return err
 				}
 				if id.String() == actorIRI.String() {
 					maybeMyFollowIRI = followId
 					break
 				}
 			}
 			// Continue breaking if we found ourselves
 			if maybeMyFollowIRI != nil {
 				break
 			}
 		}
 		// If we received an Accept whose 'object' is a Follow with an
 		// Accept that we sent, add to the following collection.
 		if maybeMyFollowIRI != nil {
 			// Verify our Follow request exists and the peer didn't
 			// fabricate it.
 			activityActors := a.GetActivityStreamsActor()
 			if activityActors == nil || activityActors.Len() == 0 {
 				return fmt.Errorf("an Accept with a Follow has no actors")
 			}
 			// This may be a duplicate check if we dereferenced the
 			// Follow above. TODO: Separate this logic to avoid
 			// redundancy.
 			//
 			// Use an anonymous function to properly scope the
 			// database lock, immediately call it.
 			err = func() error {
 				unlock, err := w.db.Lock(c, maybeMyFollowIRI)
 				if err != nil {
 					return err
 				}
 				defer unlock()
 				t, err := w.db.Get(c, maybeMyFollowIRI)
 				if err != nil {
 					return err
 				}
 				if !streams.IsOrExtendsActivityStreamsFollow(t) {
 					return fmt.Errorf("peer gave an Accept wrapping a Follow but provided a non-Follow id")
 				}
 				follow, ok := t.(Activity)
 				if !ok {
 					return fmt.Errorf("a Follow in an Accept does not satisfy the Activity interface")
 				}
 				// Ensure that we are one of the actors on the Follow.
 				ok = false
 				actors := follow.GetActivityStreamsActor()
 				for iter := actors.Begin(); iter != actors.End(); iter = iter.Next() {
 					id, err := ToId(iter)
 					if err != nil {
 						return err
 					}
 					if id.String() == actorIRI.String() {
 						ok = true
 						break
 					}
 				}
 				if !ok {
 					return fmt.Errorf("peer gave an Accept wrapping a Follow but we are not the actor on that Follow")
 				}
 				// Build map of original Accept actors
 				acceptActors := make(map[string]bool)
 				for iter := activityActors.Begin(); iter != activityActors.End(); iter = iter.Next() {
 					id, err := ToId(iter)
 					if err != nil {
 						return err
 					}
 					acceptActors[id.String()] = false
 				}
 				// Verify all actor(s) were on the original Follow.
 				followObj := follow.GetActivityStreamsObject()
 				for iter := followObj.Begin(); iter != followObj.End(); iter = iter.Next() {
 					id, err := ToId(iter)
 					if err != nil {
 						return err
 					}
 					if _, ok := acceptActors[id.String()]; ok {
 						acceptActors[id.String()] = true
 					}
 				}
 				for _, found := range acceptActors {
 					if !found {
 						return fmt.Errorf("peer gave an Accept wrapping a Follow but was not an object in the original Follow")
 					}
 				}
 				return nil
 			}()
 			if err != nil {
 				return err
 			}
 			// Add the peer to our following collection.
 			unlock, err := w.db.Lock(c, actorIRI)
 			if err != nil {
 				return err
 			}
 			// WARNING: Unlock not deferred.
 			following, err := w.db.Following(c, actorIRI)
 			if err != nil {
 				unlock()
 				return err
 			}
 			items := following.GetActivityStreamsItems()
 			if items == nil {
 				items = streams.NewActivityStreamsItemsProperty()
 				following.SetActivityStreamsItems(items)
 			}
 			for iter := activityActors.Begin(); iter != activityActors.End(); iter = iter.Next() {
 				id, err := ToId(iter)
 				if err != nil {
 					unlock()
 					return err
 				}
 				items.PrependIRI(id)
 			}
 			err = w.db.Update(c, following)
 			unlock() // unlock after regardless
 			if err != nil {
 				return err
 			}
 			// Unlock must be called by now and every branch above.
 		}
 	}
 	if w.Accept != nil {
 		return w.Accept(c, a)
 	}
 	return nil
 }
 
 // reject implements the federating Reject activity side effects.
 func (w FederatingWrappedCallbacks) reject(c context.Context, a vocab.ActivityStreamsReject) error {
 	if w.Reject != nil {
 		return w.Reject(c, a)
 	}
 	return nil
 }
 
 // add implements the federating Add activity side effects.
 func (w FederatingWrappedCallbacks) add(c context.Context, a vocab.ActivityStreamsAdd) error {
 	op := a.GetActivityStreamsObject()
 	if op == nil || op.Len() == 0 {
 		return ErrObjectRequired
 	}
 	target := a.GetActivityStreamsTarget()
 	if target == nil || target.Len() == 0 {
 		return ErrTargetRequired
 	}
 	if err := add(c, op, target, w.db); err != nil {
 		return err
 	}
 	if w.Add != nil {
 		return w.Add(c, a)
 	}
 	return nil
 }
 
 // remove implements the federating Remove activity side effects.
 func (w FederatingWrappedCallbacks) remove(c context.Context, a vocab.ActivityStreamsRemove) error {
 	op := a.GetActivityStreamsObject()
 	if op == nil || op.Len() == 0 {
 		return ErrObjectRequired
 	}
 	target := a.GetActivityStreamsTarget()
 	if target == nil || target.Len() == 0 {
 		return ErrTargetRequired
 	}
 	if err := remove(c, op, target, w.db); err != nil {
 		return err
 	}
 	if w.Remove != nil {
 		return w.Remove(c, a)
 	}
 	return nil
 }
 
 // like implements the federating Like activity side effects.
 func (w FederatingWrappedCallbacks) like(c context.Context, a vocab.ActivityStreamsLike) error {
 	op := a.GetActivityStreamsObject()
 	if op == nil || op.Len() == 0 {
 		return ErrObjectRequired
 	}
 	id, err := GetId(a)
 	if err != nil {
 		return err
 	}
 	// Create anonymous loop function to be able to properly scope the defer
 	// for the database lock at each iteration.
 	loopFn := func(iter vocab.ActivityStreamsObjectPropertyIterator) error {
 		objId, err := ToId(iter)
 		if err != nil {
 			return err
 		}
 		unlock, err := w.db.Lock(c, objId)
 		if err != nil {
 			return err
 		}
 		defer unlock()
 		if owns, err := w.db.Owns(c, objId); err != nil {
 			return err
 		} else if !owns {
 			return nil
 		}
 		t, err := w.db.Get(c, objId)
 		if err != nil {
 			return err
 		}
 		l, ok := t.(likeser)
 		if !ok {
 			return fmt.Errorf("cannot add Like to likes collection for type %T", t)
 		}
 		// Get 'likes' property on the object, creating default if
 		// necessary.
 		likes := l.GetActivityStreamsLikes()
 		if likes == nil {
 			likes = streams.NewActivityStreamsLikesProperty()
 			l.SetActivityStreamsLikes(likes)
 		}
 		// Get 'likes' value, defaulting to a collection.
 		likesT := likes.GetType()
 		if likesT == nil {
 			col := streams.NewActivityStreamsCollection()
 			likesT = col
 			likes.SetActivityStreamsCollection(col)
 		}
 		// Prepend the activity's 'id' on the 'likes' Collection or
 		// OrderedCollection.
 		if col, ok := likesT.(itemser); ok {
 			items := col.GetActivityStreamsItems()
 			if items == nil {
 				items = streams.NewActivityStreamsItemsProperty()
 				col.SetActivityStreamsItems(items)
 			}
 			items.PrependIRI(id)
 		} else if oCol, ok := likesT.(orderedItemser); ok {
 			oItems := oCol.GetActivityStreamsOrderedItems()
 			if oItems == nil {
 				oItems = streams.NewActivityStreamsOrderedItemsProperty()
 				oCol.SetActivityStreamsOrderedItems(oItems)
 			}
 			oItems.PrependIRI(id)
 		} else {
 			return fmt.Errorf("likes type is neither a Collection nor an OrderedCollection: %T", likesT)
 		}
 		err = w.db.Update(c, t)
 		if err != nil {
 			return err
 		}
 		return nil
 	}
 	for iter := op.Begin(); iter != op.End(); iter = iter.Next() {
 		if err := loopFn(iter); err != nil {
 			return err
 		}
 	}
 	if w.Like != nil {
 		return w.Like(c, a)
 	}
 	return nil
 }
 
 // announce implements the federating Announce activity side effects.
 func (w FederatingWrappedCallbacks) announce(c context.Context, a vocab.ActivityStreamsAnnounce) error {
 	id, err := GetId(a)
 	if err != nil {
 		return err
 	}
 	op := a.GetActivityStreamsObject()
 	// Create anonymous loop function to be able to properly scope the defer
 	// for the database lock at each iteration.
 	loopFn := func(iter vocab.ActivityStreamsObjectPropertyIterator) error {
 		objId, err := ToId(iter)
 		if err != nil {
 			return err
 		}
 		unlock, err := w.db.Lock(c, objId)
 		if err != nil {
 			return err
 		}
 		defer unlock()
 		if owns, err := w.db.Owns(c, objId); err != nil {
 			return err
 		} else if !owns {
 			return nil
 		}
 		t, err := w.db.Get(c, objId)
 		if err != nil {
 			return err
 		}
 		s, ok := t.(shareser)
 		if !ok {
 			return fmt.Errorf("cannot add Announce to Shares collection for type %T", t)
 		}
 		// Get 'shares' property on the object, creating default if
 		// necessary.
 		shares := s.GetActivityStreamsShares()
 		if shares == nil {
 			shares = streams.NewActivityStreamsSharesProperty()
 			s.SetActivityStreamsShares(shares)
 		}
 		// Get 'shares' value, defaulting to a collection.
 		sharesT := shares.GetType()
 		if sharesT == nil {
 			col := streams.NewActivityStreamsCollection()
 			sharesT = col
 			shares.SetActivityStreamsCollection(col)
 		}
 		// Prepend the activity's 'id' on the 'shares' Collection or
 		// OrderedCollection.
 		if col, ok := sharesT.(itemser); ok {
 			items := col.GetActivityStreamsItems()
 			if items == nil {
 				items = streams.NewActivityStreamsItemsProperty()
 				col.SetActivityStreamsItems(items)
 			}
 			items.PrependIRI(id)
 		} else if oCol, ok := sharesT.(orderedItemser); ok {
 			oItems := oCol.GetActivityStreamsOrderedItems()
 			if oItems == nil {
 				oItems = streams.NewActivityStreamsOrderedItemsProperty()
 				oCol.SetActivityStreamsOrderedItems(oItems)
 			}
 			oItems.PrependIRI(id)
 		} else {
 			return fmt.Errorf("shares type is neither a Collection nor an OrderedCollection: %T", sharesT)
 		}
 		err = w.db.Update(c, t)
 		if err != nil {
 			return err
 		}
 		return nil
 	}
 	if op != nil {
 		for iter := op.Begin(); iter != op.End(); iter = iter.Next() {
 			if err := loopFn(iter); err != nil {
 				return err
 			}
 		}
 	}
 	if w.Announce != nil {
 		return w.Announce(c, a)
 	}
 	return nil
 }
 
 // undo implements the federating Undo activity side effects.
 func (w FederatingWrappedCallbacks) undo(c context.Context, a vocab.ActivityStreamsUndo) error {
 	op := a.GetActivityStreamsObject()
 	if op == nil || op.Len() == 0 {
 		return ErrObjectRequired
 	}
 	actors := a.GetActivityStreamsActor()
 	if err := mustHaveActivityActorsMatchObjectActors(c, actors, op, w.newTransport, w.inboxIRI); err != nil {
 		return err
 	}
 	if w.Undo != nil {
 		return w.Undo(c, a)
 	}
 	return nil
 }
 
 // block implements the federating Block activity side effects.
 func (w FederatingWrappedCallbacks) block(c context.Context, a vocab.ActivityStreamsBlock) error {
 	op := a.GetActivityStreamsObject()
 	if op == nil || op.Len() == 0 {
 		return ErrObjectRequired
 	}
 	if w.Block != nil {
 		return w.Block(c, a)
 	}
 	return nil
 }