processing.go

package processing

import (
	"context"
	"crypto"
	"time"

	"git.sr.ht/~mariusor/lw"
	vocab "github.com/go-ap/activitypub"
	"github.com/go-ap/errors"
)

type ActivityPubClient interface {
	CtxToCollection(context.Context, vocab.Item, ...vocab.IRI) (vocab.IRI, vocab.Item, error)
	CtxLoadIRI(context.Context, vocab.IRI) (vocab.Item, error)
}

type P struct {
	baseIRI vocab.IRIs
	c       ActivityPubClient
	s       Store
	l       lw.Logger

	retries int
	sync    bool

	// skipValidationOnInboundCollections determines if the validation functionality checks that the collection
	// which received the activity actually exists.
	skipValidationOnInboundCollections bool

	// localIRICheckFn is a function that can be passed from outside the module to determine if a [vocab.IRI] "is local".
	// This usually means that the storage layer can dereference the IRI to an object that is stored locally.
	localIRICheckFn IRIValidator

	// createIDFn is a function that can be used to generate IDs for the ActivityPub objects that get created during the
	// processing of **client** activities.
	createIDFn IDGenerator

	// actorKeyGenFn is a function that can be used to generate the public and private key pairs for actors that get
	// created during the processing of **client** activities.
	actorKeyGenFn vocab.WithActorFn

	timeNowFn TimeNow
}

type emptyClient struct{}

func (d emptyClient) CtxToCollection(ctx context.Context, item vocab.Item, iri ...vocab.IRI) (vocab.IRI, vocab.Item, error) {
	return "", nil, errors.NotImplementedf("invalid http.client set on the processor")
}

func (d emptyClient) CtxLoadIRI(ctx context.Context, iri vocab.IRI) (vocab.Item, error) {
	return nil, errors.NotImplementedf("invalid http.client set on the processor")
}

var (
	nilLogger = lw.Nil()
	nilClient = emptyClient{}
)

func New(o ...OptionFn) *P {
	p := P{
		l:               nilLogger,
		c:               nilClient,
		retries:         DefaultRetryCount,
		createIDFn:      emptyIDGenerator,
		localIRICheckFn: defaultLocalIRICheck,
		actorKeyGenFn:   defaultKeyGenerator,
		timeNowFn:       defaultTimeNowFn,
	}
	for _, fn := range o {
		fn(&p)
	}
	return &p
}

type OptionFn func(s *P)

// WithSynchronousProcessing configures the processor to not execute any of the potential asynchronous processing steps
// in an asynchronous manner.
func WithSynchronousProcessing(p *P) {
	p.sync = true
}

// WithInboundCollectionValidationSkip configures the processor to skip checking that the collection in which the
// activity that's being processed has been received, actually exists.
func WithInboundCollectionValidationSkip(p *P) {
	p.skipValidationOnInboundCollections = true
}

// WithDisseminationRetryCount specifies the number of retries for failed remote activity dissemination.
// The default value is 0 when building with -tags dev, and 5 otherwise.
// If a negative value is passed as the retry count, we don't try to execute the dissemination at all.
func WithDisseminationRetryCount(r int) OptionFn {
	return func(p *P) {
		p.retries = r
	}
}

// WithIDGenerator specifies the function that generates IDs for Activities and Objects handled in an Outbox.
func WithIDGenerator(genFn IDGenerator) OptionFn {
	return func(p *P) {
		if genFn != nil {
			p.createIDFn = genFn
		}
	}
}

// WithActorKeyGenerator specifies the function that generates a Private and Public key pair for an Actor
// that was created as a side effect of a Create Activity in an Outbox.
func WithActorKeyGenerator(genFn vocab.WithActorFn) OptionFn {
	return func(p *P) {
		if genFn != nil {
			p.actorKeyGenFn = genFn
		}
	}
}

// WithLogger sets a logger for the processor.
func WithLogger(l lw.Logger) OptionFn {
	return func(p *P) {
		p.l = l
	}
}

// WithClient sets an ActivityPub client on the processor.
// It will be used to dereference remote IRIs that need a full representation, and also for disseminating
// Activities to other servers.
// In order to be able to perform the later it needs to be configured with a valid HTTP-Signatures authorization
// mechanism.
func WithClient(c ActivityPubClient) OptionFn {
	return func(p *P) {
		p.c = c
	}
}

// WithStorage sets the local storage for the processor.
func WithStorage(s Store) OptionFn {
	return func(p *P) {
		p.s = s
	}
}

// WithIRI configures the IRI list that the processor considers as being local.
func WithIRI(i ...vocab.IRI) OptionFn {
	return func(p *P) {
		p.baseIRI = i
	}
}

// WithLocalIRIChecker configures an external function that is used to determine if an IRI is local.
func WithLocalIRIChecker(isLocalFn IRIValidator) OptionFn {
	return func(p *P) {
		if isLocalFn != nil {
			p.localIRICheckFn = isLocalFn
		}
	}
}

// WithTimeFn configures an external function to generate the Published, Updated, Deleted times for
// Activities and Objects created as side effects of Content Management Outbox operations.
func WithTimeFn(timeNowFn TimeNow) OptionFn {
	return func(p *P) {
		if timeNowFn != nil {
			p.timeNowFn = timeNowFn
		}
	}
}

func validateLocalIRI(s Store, iris ...vocab.IRI) vocab.IRIs {
	if s == nil {
		return iris
	}
	validIRIs := make(vocab.IRIs, 0, len(iris))
	for _, iri := range iris {
		it, err := s.Load(iri)
		if err == nil && !vocab.IsNil(it) {
			_ = validIRIs.Append(iri)
		}
	}
	return validIRIs
}

// ProcessActivity processes the "it" Activity received in the "receivedIn" collection.
// The "author" parameter is the Actor that has been determined to be the Activity's author.
func (p P) ProcessActivity(it vocab.Item, author vocab.Actor, receivedIn vocab.IRI) (vocab.Item, error) {
	if vocab.IsNil(it) {
		return nil, InvalidActivity("received nil")
	}
	p.l = p.l.WithContext(lw.Ctx{"in": receivedIn, "type": it.GetType(), "author": author.ID})
	p.l.Debugf("Started")
	defer func(start time.Time) {
		p.l.WithContext(lw.Ctx{"duration": time.Since(start)}).Debugf("Ended")
	}(time.Now().Round(0))

	switch {
	case IsOutbox(receivedIn):
		return p.ProcessClientActivity(it, author, receivedIn)
	case IsInbox(receivedIn):
		return p.ProcessServerActivity(it, author, receivedIn)
	default:
		return nil, errors.MethodNotAllowedf("unable to process activities at current IRI: %s", receivedIn)
	}
}

func (p *P) createNewTags(tags vocab.Item, parent vocab.Item) error {
	if vocab.IsNil(tags) {
		return nil
	}
	// According to the example in the Implementation Notes on the Activity Streams Vocabulary spec,
	// tag objects are ActivityStreams Objects without a type, that's why we use an empty string valid type:
	// https://www.w3.org/TR/activitystreams-vocabulary/#microsyntaxes
	validTagTypes := vocab.ActivityVocabularyTypes{vocab.MentionType, vocab.ObjectType, vocab.NilType}
	_ = vocab.OnItem(tags, func(tag vocab.Item) error {
		if validTagTypes.Match(tag.GetType()) {
			return nil
		}
		if id := tag.GetID(); len(id) > 0 {
			return nil
		}
		if err := SetIDIfMissing(tag, parent, p.createIDFn); err == nil {
			_, _ = p.s.Save(tag)
		}
		return nil
	})
	return nil
}

func loadBlockedForActor(loader ReadStore, rec vocab.Item) vocab.ItemCollection {
	blockedCol, err := loader.Load(BlockedCollection.IRI(rec))
	if err != nil {
		return nil
	}

	var blocked vocab.ItemCollection
	_ = vocab.OnCollectionIntf(blockedCol, func(c vocab.CollectionInterface) error {
		blocked = c.Collection()
		return nil
	})
	return blocked
}

func (p *P) actorHasBlockedFn(rec vocab.Item) func(act vocab.Item) bool {
	blocked := loadBlockedForActor(p.s, rec)
	if blocked == nil {
		return func(_ vocab.Item) bool {
			return false
		}
	}

	return func(act vocab.Item) bool {
		result := false
		_ = vocab.OnItem(act, func(item vocab.Item) error {
			if result = blocked.IRIs().Contains(item.GetID()); result {
				return errors.Newf("skip")
			}
			return nil
		})
		return result
	}
}

type KeyLoader interface {
	LoadKey(vocab.IRI) (crypto.PrivateKey, error)
}

// BuildReplyToCollections builds the list of objects that it is inReplyTo
func (p *P) BuildReplyToCollections(it vocab.Item) vocab.ItemCollection {
	ob, err := vocab.ToObject(it)
	if err != nil {
		return nil
	}
	if vocab.IsNil(ob.InReplyTo) {
		return nil
	}

	collections := make(vocab.ItemCollection, 0)
	_ = vocab.OnItem(ob.InReplyTo, func(replyTo vocab.Item) error {
		if !p.IsLocal(replyTo) {
			return nil
		}
		return collections.Append(vocab.Replies.IRI(replyTo.GetLink()))
	})
	return collections
}

// CollectionManagementActivity processes matching activities
//
// https://www.w3.org/TR/activitystreams-vocabulary/#h-motivations-collections
//
// The Collection Management use case primarily deals with activities involving the management of content within collections.
// Examples of collections include things like folders, albums, friend lists, etc.
// This includes, for instance, activities such as "Sally added a file to Folder A",
// "John moved the file from Folder A to Folder B", etc.
func (p *P) CollectionManagementActivity(act *vocab.Activity) (*vocab.Activity, error) {
	if vocab.IsNil(act.Object) {
		return act, InvalidActivityObject("is nil for %T[%s]", act, act.GetType())
	}
	switch {
	case vocab.AddType.Match(act.Type):
		return p.AddActivity(act)
	case vocab.MoveType.Match(act.Type):
		return p.MoveActivity(act)
	case vocab.RemoveType.Match(act.Type):
		return p.RemoveActivity(act)
	default:
	}
	return act, errors.BadRequestf("Invalid type %v", act.GetType())
}

// EventRSVPActivity processes matching activities
//
// https://www.w3.org/TR/activitystreams-vocabulary/#h-motivations-rsvp
//
// The Event RSVP use case primarily deals with invitations to events and RSVP type responses.
func EventRSVPActivity(l WriteStore, act *vocab.Activity) (*vocab.Activity, error) {
	if vocab.IsNil(act.Object) {
		return act, InvalidActivityObject("is nil for %T[%s]", act, act.GetType())
	}
	switch {
	case vocab.AcceptType.Match(act.Type):
	case vocab.IgnoreType.Match(act.Type):
	case vocab.InviteType.Match(act.Type):
	case vocab.RejectType.Match(act.Type):
	case vocab.TentativeAcceptType.Match(act.Type):
	case vocab.TentativeRejectType.Match(act.Type):
	default:
		return nil, errors.BadRequestf("Invalid type %v", act.GetType())
	}
	return act, errors.NotImplementedf("Processing %s activity is not implemented", act.GetType())
}

// GroupManagementActivity processes matching activities
//
// https://www.w3.org/TR/activitystreams-vocabulary/#h-motivations-group
//
// The Group Management use case primarily deals with management of groups.
// It can include, for instance, activities such as "John added Sally to Group A", "Sally joined Group A",
// "Joe left Group A", etc.
func GroupManagementActivity(l WriteStore, act *vocab.Activity) (*vocab.Activity, error) {
	// TODO(marius):
	return act, errors.NotImplementedf("Processing %s activity is not implemented", act.GetType())
}

// ContentExperienceActivity processes matching activities
//
// https://www.w3.org/TR/activitystreams-vocabulary/#h-motivations-experience
//
// The Content Experience use case primarily deals with describing activities involving listening to,
// reading, or viewing content. For instance, "Sally read the article", "Joe listened to the song".
func ContentExperienceActivity(l WriteStore, act *vocab.Activity) (*vocab.Activity, error) {
	// TODO(marius):
	return act, errors.NotImplementedf("Processing %s activity is not implemented", act.GetType())
}

// GeoSocialEventsActivity processes matching activities
//
// https://www.w3.org/TR/activitystreams-vocabulary/#h-motivations-geo
//
// The Geo-Social Events use case primarily deals with activities involving geo-tagging type activities. For instance,
// it can include activities such as "Joe arrived at work", "Sally left work", and "John is travel from home to work".
func GeoSocialEventsActivity(l WriteStore, act *vocab.Activity) (*vocab.Activity, error) {
	// TODO(marius):
	return act, errors.NotImplementedf("Processing %s activity is not implemented", act.GetType())
}

// GeoSocialEventsIntransitiveActivity processes matching activities
//
// https://www.w3.org/TR/activitystreams-vocabulary/#h-motivations-geo
//
// The Geo-Social Events use case primarily deals with activities involving geo-tagging type activities. For instance,
// it can include activities such as "Joe arrived at work", "Sally left work", and "John is travel from home to work".
func GeoSocialEventsIntransitiveActivity(l WriteStore, act vocab.Item) (vocab.Item, error) {
	// TODO(marius):
	return act, errors.NotImplementedf("Processing %s activity is not implemented", act.GetType())
}

// OffersActivity processes matching activities
//
// https://www.w3.org/TR/activitystreams-vocabulary/#h-motivations-offer
//
// The Offers use case deals with activities involving offering one object to another. It can include, for instance,
// activities such as "Company A is offering a discount on purchase of Product Z to Sally",
// "Sally is offering to add a File to Folder A", etc.
func OffersActivity(l WriteStore, act *vocab.Activity) (*vocab.Activity, error) {
	// TODO(marius):
	return act, errors.NotImplementedf("Processing %s activity is not implemented", act.GetType())
}