cmd-activitypub.go

package fedbox

import (
	"bytes"
	"fmt"
	"io"
	"net/url"
	"os"
	"slices"
	"time"

	vocab "github.com/go-ap/activitypub"
	"github.com/go-ap/errors"
	ap "github.com/go-ap/fedbox/activitypub"
	"github.com/go-ap/filters"
)

type AddActorCmd struct {
	Type         vocab.ActivityVocabularyType `help:"The type of the ActivityPub actor to add."`
	KeyType      string                       `help:"Type of keys to generate: ${keyTypes}" enum:"${keyTypes}" default:"${defaultKeyType}"`
	AttributedTo vocab.IRI                    `help:"The IRI of the Actor we should use as author."`
	Tags         []string                     `name:"tag" help:"The tag(s) to attach to the actor."`
	Names        []string                     `arg:"" name:"name" help:"The name(s) of the actor."`
}

func (a AddActorCmd) Run(ctl *Base) error {
	keyType := a.KeyType
	if len(a.Names) == 0 {
		// FIXME(marius): pass stdIn also to the Run
		name, err := loadFromStdin("Enter the actor's name")
		if err != nil {
			return errors.Errorf("Missing the actor's name")
		}
		a.Names = append(a.Names, string(name))
	}

	author := ap.Self(vocab.IRI(ctl.Conf.BaseURL), AppName)
	if authIRI := a.AttributedTo; len(authIRI) > 0 {
		act, err := ap.LoadActor(ctl.Storage, authIRI)
		if err != nil {
			return err
		}
		author = act
	}

	tags := make(vocab.ItemCollection, 0)

	objectsCollection := filters.ObjectsType.IRI(vocab.IRI(ctl.Conf.BaseURL))
	allObjects, _ := ctl.Storage.Load(objectsCollection)
	_ = vocab.OnItem(allObjects, func(it vocab.Item) error {
		return vocab.OnObject(it, func(object *vocab.Object) error {
			for _, tag := range a.Tags {
				if object.Name.First().String() != tag {
					continue
				}
				if object.AttributedTo.GetLink() != author.GetLink() {
					continue
				}
				_ = tags.Append(object)
			}
			return nil
		})
	})

	rw := muxReadWriter{Reader: ctl.in, Writer: ctl.out}
	actors := make(vocab.ItemCollection, 0)
	for _, name := range a.Names {
		pw, err := loadPwFromStdin(rw, fmt.Sprintf("%s's password: ", name))
		if err != nil {
			return err
		}
		if !vocab.ActorTypes.Match(a.Type) {
			a.Type = vocab.PersonType
		}

		now := time.Now().UTC()
		p := &vocab.Actor{
			Type: a.Type,
			// TODO(marius): when adding authentication to the command, we can set here the actor that executes it
			AttributedTo:      author.GetLink(),
			Generator:         ctl.Service.ID,
			Published:         now,
			PreferredUsername: vocab.DefaultLangValue(name),
		}
		if len(tags) > 0 {
			p.Tag = tags
		}
		id, err := GenerateID(ctl.Service.ID)(p, author)
		if err == nil {
			p.Inbox = vocab.Inbox.IRI(id)
			p.Outbox = vocab.Outbox.IRI(id)
			p.Followers = vocab.Followers.IRI(id)
			p.Following = vocab.Following.IRI(id)

			oauth := id.AddPath("oauth")
			proxyURL := ctl.Service.ID.AddPath("proxyUrl")
			p.Endpoints = &vocab.Endpoints{
				OauthAuthorizationEndpoint: oauth.AddPath("authorize"),
				OauthTokenEndpoint:         oauth.AddPath("token"),
				ProxyURL:                   proxyURL,
				SharedInbox:                vocab.Inbox.IRI(ctl.Service),
			}
		}

		if p, err = ctl.AddActor(p, &author, false); err != nil {
			return err
		}
		_, _ = fmt.Fprintf(ctl.out, "Added %q %q\n", a.Type, name)
		_, _ = fmt.Fprintf(ctl.out, "\t%s\n", p.GetLink())
		pair, _ := ap.GenerateKeyPair(ap.KeyType(keyType))
		if pair != nil {
			if err := ap.AddKeyToItem(ctl.Storage, p, *pair); err != nil {
				Errf(ctl.err, "Error saving metadata for %s: %s", name, err)
			}
		}
		if pw != nil {
			err = ctl.Storage.PasswordSet(p.ID, pw)
		}
		_ = actors.Append(p)
	}
	return nil
}

type ActorsCmd struct {
	Add AddActorCmd `cmd:"" help:"Adds an ActivityPub actor."`
}

type Pub struct {
	Actors ActorsCmd `cmd:"" name:"actor" help:"Actor management helper."`
	Add    AddCmd    `cmd:"" name:"add" help:"Adds a new object."`
	List   ListCmd   `cmd:"" help:"Lists objects."`
	Info   InfoCmd   `cmd:"" help:"Show information about an object."`
	Delete DeleteCmd `cmd:"" help:"Deletes an ActivityPub object."`
	Move   MoveCmd   `cmd:"" help:"Move ActivityPub objects to a new collection."`
	Copy   CopyCmd   `cmd:"" help:"Copy ActivityPub objects."`
	Index  IndexCmd  `cmd:"" help:"Reindex current storage ActivityPub objects."`
	Export ExportCmd `cmd:"" help:"Exports ActivityPub objects."`
	Import ImportCmd `cmd:"" help:"Imports ActivityPub objects."`
}

var ValidGenericTypes = vocab.ActivityVocabularyTypes{vocab.ObjectType, vocab.ActorType}

type DeleteCmd struct {
	Reason    string      `help:"The reason why we want to delete the item"`
	InReplyTo []string    `help:"If deletion is a followup on moderation activities"`
	IRIs      []vocab.IRI `arg:"" name:"iris"`
}

func (d DeleteCmd) Run(ctl *Base) error {
	return ctl.DeleteObjects(d.Reason, d.InReplyTo, d.IRIs...)
}

type ListCmd struct {
	Type   []vocab.ActivityVocabularyType `help:"The type of ActivityPub object to list" default:"${defaultObjectTypes}"`
	Output string                         `help:"The format in which to output the items." enum:"text,json" default:"text"`
	IRIs   []vocab.IRI                    `arg:"" name:"iris"`
}

func (l ListCmd) Run(ctl *Base) error {
	typeFl := l.Type

	var paths vocab.IRIs
	if len(l.IRIs) == 0 {
		paths = append(paths,
			filters.ObjectsType.IRI(vocab.IRI(ctl.Conf.BaseURL)),
			filters.ActorsType.IRI(vocab.IRI(ctl.Conf.BaseURL)),
			filters.ActivitiesType.IRI(vocab.IRI(ctl.Conf.BaseURL)),
		)
	} else {
		for _, path := range l.IRIs {
			u, err := url.ParseRequestURI(path.String())
			if err != nil {
				continue
			}
			if u.Host == "" {
				u.Host = ctl.Conf.BaseURL
			}
			paths = append(paths, vocab.IRI(u.String()))
		}
	}

	all, err := ctl.List(paths, typeFl...)
	if err != nil {
		return err
	}
	slices.SortStableFunc(all, filters.TimestampSortFunc)
	_ = printItem(ctl.out, all, l.Output)
	return nil
}

type AddCmd struct {
	Type         vocab.ActivityVocabularyType `help:"The type of ActivityPub object(s) to create." default:"${defaultObjectTypes}"`
	Name         string                       `help:"The name of the ActivityPub object(s) to create."`
	AttributedTo string                       `help:"The IRI of the Actor we should use as author"`
	Tag          []string                     `help:"The tag(s) to attach to the object."`
}

var validObjects = append(vocab.ObjectTypes, vocab.ObjectType, "")

func (a AddCmd) Run(ctl *Base) error {
	f := make(filters.Checks, 0)

	incType := a.Type
	if !validObjects.Match(incType) {
		return errors.Errorf("This command only supports only object of types %v", vocab.ObjectTypes)
	} else {
		incType = vocab.NoteType
	}
	prop := "title"
	if vocab.ActorTypes.Match(incType) {
		prop = "name"
	}

	if typ := types(incType); typ != nil {
		f = append(f, typ)
	}

	incName := a.Name
	if len(incName) == 0 {
		if n, err := loadFromStdin("Enter the %s's %s", incType, prop); err == nil {
			incName = string(n)
		}
	}
	if n := names(incName); n != nil {
		f = append(f, n)
	}

	authIRI := vocab.IRI(a.AttributedTo)
	if len(authIRI) == 0 {
		authIRI = vocab.IRI(ctl.Conf.BaseURL)
	}
	author, err := ap.LoadActor(ctl.Storage, authIRI, f...)
	if err != nil {
		return err
	}

	now := time.Now().UTC()
	p := &vocab.Object{
		Type: incType,
		// TODO(marius): when adding authentication to the command, we can set here the actor that executes it
		AttributedTo: author.GetLink(),
		CC:           vocab.ItemCollection{vocab.PublicNS},
		Published:    now,
		Name:         vocab.DefaultLangValue(incName),
	}

	if p, err = ctl.AddObject(p, author); err != nil {
		return errors.Annotatef(err, "Unable to save object")
	}
	_, _ = fmt.Fprintf(ctl.out, "Added %s [%s]: %s\n", incType, incName, p.GetLink())

	return nil
}

type ImportCmd struct {
	Base        vocab.IRI  `flag:"" help:"The base IRI to replace"`
	SkipRemotes bool       `flag:"" help:"Do not try to disseminate to remote recipients"`
	Files       []*os.File `arg:"" help:"The files containing the JSON encoded items."`
}

var activityTypes = append(vocab.ActivityTypes, vocab.IntransitiveActivityTypes...)

func (i ImportCmd) Run(ctl *Base) error {
	baseIRI := ctl.Conf.BaseURL
	toReplace := i.Base

	var col vocab.ItemCollection
	for _, f := range i.Files {
		buf, err := io.ReadAll(f)
		if err != nil {
			Errf(ctl.err, "Error %v", err)
			continue
		}
		if len(buf) == 0 {
			Errf(ctl.err, "Empty file %s", f.Name())
			continue
		}

		if len(toReplace) > 0 {
			buf = bytes.Replace(buf, []byte(toReplace), []byte(baseIRI), -1)
		}
		ob, err := vocab.UnmarshalJSON(buf)
		if err != nil {
			Errf(ctl.err, "Error unmarshaling JSON: %+v", err)
			continue
		}

		if vocab.IsItemCollection(ob) {
			var isItThough bool
			if col, isItThough = ob.(vocab.ItemCollection); !isItThough {
				Errf(ctl.err, "Invalid item collection unmarshaled: %+v", err)
			}
		} else {
			col = vocab.ItemCollection{ob}
		}
	}

	for i, it := range col {
		typ := it.GetType()
		if !activityTypes.Match(typ) {
			// NOTE(marius): we wrap single objects into Create activities in order
			// to get them handled through the ProcessingActivity functionality which
			// enhances the object with various elements.
			if exists, err := ctl.Storage.Load(it.GetLink()); errors.IsNotFound(err) || vocab.IsNil(exists) {
				it = ap.WrapObjectInCreate(it, ctl.Service)
			} else {
				it = ap.WrapObjectInUpdate(it, ctl.Service)
			}
		}
		col[i] = it
	}

	slices.SortStableFunc(col, SortImportActivitiesByDeps)

	ObjectTypeOf := func(it vocab.Item) (typ vocab.Typer) {
		vocab.OnActivity(it, func(act *vocab.Activity) error {
			if act.Object != nil {
				typ = act.Object.GetType()
			}
			return nil
		})
		return typ
	}
	count := 0
	start := time.Now().Round(0)
	for _, it := range col {
		_, _ = fmt.Fprintf(ctl.out, "Processing %s[%v]\n", it.GetType(), ObjectTypeOf(it))
		actor := &ctl.Service

		importer := ctl.Saver(&ctl.Service, i.SkipRemotes, true)
		err := vocab.OnIntransitiveActivity(it, func(a *vocab.IntransitiveActivity) (err error) {
			if a == nil {
				Errf(ctl.err, "invalid activity, is nil")
				return nil
			}
			if vocab.IsIRI(a.Actor) {
				if maybeActor, err := importer.DereferenceItem(a.Actor); err == nil {
					if act, err := vocab.ToActor(maybeActor); err == nil {
						actor = act
						a.Actor = act
					}
				}
				return nil
			}
			return err
		})
		if err != nil {
			continue
		}

		it, err = importer.ProcessClientActivity(it, *actor, vocab.Outbox.IRI(actor))
		if err != nil {
			Errf(ctl.err, "Unable to process %s %s: %v", it.GetType(), it.GetID(), err)
			continue
		}
		count++
	}

	tot := time.Now().Sub(start)
	_, _ = fmt.Fprintf(ctl.out, "Activities count:             % 10d\n", len(col))
	_, _ = fmt.Fprintf(ctl.out, "Activities processing time:   %10s\n", tot.Round(500*time.Microsecond))
	if count > 0 {
		perIt := time.Duration(int64(tot) / int64(count))
		_, _ = fmt.Fprintf(ctl.out, "Elapsed time per activity:    %10s\n", perIt.Round(500*time.Microsecond))
	}

	_, _ = fmt.Fprintf(ctl.out, "Import done!\n")
	return nil
}

// SortImportActivitiesByDeps can be used as a [slices.SortFn] to
// sort an activity collection in a consistent order for import,
// where creating actors comes first, creating objects comes second,
// the others activities at the end.
func SortImportActivitiesByDeps(c1, c2 vocab.Item) int {
	const (
		switchPos = -1
		keepPos   = 1
		meh       = 0
	)

	t1 := vocab.CreateType.Match(c1.GetType())
	t2 := vocab.CreateType.Match(c2.GetType())

	// NOTE(marius): if first item is a Create activity it remains before
	if t1 && !t2 {
		return 100 * switchPos
	} else if t2 && !t1 {
		return 100 * keepPos
	} else if !t1 && !t2 {
		return meh
	}

	// NOTE(marius): both items are create activities
	actObFn := func(a vocab.Item) (vocab.Item, vocab.Item) {
		var auth vocab.Item
		var ob vocab.Item
		_ = vocab.OnActivity(a, func(act *vocab.Activity) error {
			auth = act.Actor
			ob = act.Object
			return nil
		})
		return auth, ob
	}

	act1, ob1 := actObFn(c1)
	act2, ob2 := actObFn(c2)

	at1 := vocab.ActorTypes.Match(ob1.GetType())
	at2 := vocab.ActorTypes.Match(ob2.GetType())

	// NOTE(marius): check if one Create object is an actor (it should be at top)
	if at1 && !at2 {
		return 10 * switchPos
	} else if at2 && !at1 {
		return 10 * keepPos
	} else if !at1 && !at2 {
		return meh
	}

	if ob1.GetID().Equal(act2.GetID()) {
		return 5 * switchPos
	}
	if ob2.GetID().Equal(act1.GetID()) {
		return 5 * keepPos
	}

	authFn := func(o vocab.Item) vocab.IRI {
		var auth vocab.IRI
		_ = vocab.OnObject(o, func(ob *vocab.Object) error {
			if ob.AttributedTo != nil {
				auth = ob.AttributedTo.GetLink()
			}
			return nil
		})
		return auth
	}

	// NOTE(marius): for Actor Create activities we check that the actor's attributedTo.
	auth1 := authFn(ob1)
	auth2 := authFn(ob2)
	if vocab.IsNil(auth1) || vocab.IsNil(auth2) || auth1.Equal(auth2) {
		return meh
	}

	if act2.GetID().Equal(auth1.GetID()) {
		return 2 * switchPos
	} else if act1.GetID().Equal(auth2.GetID()) {
		return 2 * keepPos
	}

	return meh
}

func checkForTypeMatcher(types ...vocab.ActivityVocabularyType) func(t1, t2 vocab.Typer) (int, bool) {
	return func(t1, t2 vocab.Typer) (int, bool) {
		if vocab.ActivityVocabularyTypes(types).Match(t1) {
			if vocab.ActivityVocabularyTypes(types).Match(t2) {
				return 0, true
			}
			return -1, true
		} else if vocab.ActivityVocabularyTypes(types).Match(t2) {
			return 1, true
		}
		return -2, false
	}
}

func checkActorTypes(types ...vocab.ActivityVocabularyType) func(c1, c2 vocab.Item) int {
	return func(c1, c2 vocab.Item) int {
		a1, _ := vocab.ToActivity(c1)
		a2, _ := vocab.ToActivity(c2)
		if a1 == nil {
			return -1
		}
		if a2 == nil {
			return 1
		}
		st, _ := checkForTypeMatcher(types...)(a1.Actor.GetType(), a2.Actor.GetType())
		return st
	}
}

type ExportCmd struct {
	To string `name:"to" optional:"" help:"The files where to output the items, if absent it will be printed to stdout."`
}

func toItemCol(it vocab.Item) vocab.ItemCollection {
	col := make(vocab.ItemCollection, 0)
	_ = vocab.OnItem(it, func(item vocab.Item) error {
		return col.Append(item)
	})
	return col
}

func dumpAll(ctl *Base, iri vocab.IRI, f ...filters.Check) (vocab.ItemCollection, error) {
	col := make(vocab.ItemCollection, 0)

	objects, err := ctl.Storage.Load(iri, f...)
	if err != nil {
		return col, err
	}

	return col, col.Append(toItemCol(objects)...)
}

func (e ExportCmd) Run(ctl *Base) error {
	baseURL := vocab.IRI(ctl.Conf.BaseURL)
	allCollections := vocab.CollectionPaths{filters.ActivitiesType, filters.ActorsType, filters.ObjectsType}
	objects := make(vocab.ItemCollection, 0, 4)
	for _, col := range allCollections {
		dump, err := dumpAll(ctl, vocab.IRIf(baseURL, col))
		if err != nil {
			return err
		}
		_ = objects.Append(dump...)
	}
	if len(objects) == 0 {
		return errors.Errorf("No objects to export")
	}
	slices.SortStableFunc(objects, filters.TimestampSortFunc)
	for i, it := range objects {
		objects[i] = vocab.FlattenProperties(it)
	}
	where := os.Stdout
	if e.To != "" {
		f, err := os.OpenFile(e.To, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0600)
		if err != nil {
			return err
		}
		defer f.Close()
		where = f
	}
	return OutJSON(where)(objects)
}

type InfoCmd struct {
	IRIs   []vocab.IRI `arg:"" name:"iris"`
	Output string      `help:"The format in which to output the items." enum:"text,json" default:"text"`
}

func (l InfoCmd) Run(ctl *Base) error {
	objects := make(vocab.ItemCollection, 0, len(l.IRIs))
	if len(l.IRIs) == 0 {
		return errors.Errorf("No IRIs passed")
	}
	for _, iri := range l.IRIs {
		ob, err := ctl.Storage.Load(iri)
		if err != nil {
			Errf(ctl.err, err.Error())
			continue
		}
		_ = objects.Append(ob)
	}

	for _, ob := range objects {
		_ = printItem(ctl.out, ob, l.Output)
	}
	return nil
}

type MoveCmd struct {
	To   vocab.IRI   `help:"Collection to which to move the objects." required:""`
	IRIs []vocab.IRI `arg:"" name:"iris"`
}

func (m MoveCmd) Run(ctl *Base) error {
	return ctl.MoveObjects(m.To, m.IRIs...)
}

type CopyCmd struct {
	To   vocab.IRI   `help:"Collection to which to move the objects." required:""`
	IRIs []vocab.IRI `arg:"" name:"iris"`
}

func (c CopyCmd) Run(ctl *Base) error {
	return ctl.CopyObjects(c.To, c.IRIs...)
}

type IndexCmd struct{}

type reindexer interface {
	Reindex() error
}

func (i IndexCmd) Run(ctl *Base) error {
	start := time.Now()

	indexer, ok := ctl.Storage.(reindexer)
	if !ok {
		_, _ = fmt.Fprintf(os.Stderr, "Current storage engine %T does not support reindexing\n", ctl.Storage)
		return errors.Newf("unsupported storage engine")
	}

	if err := indexer.Reindex(); err != nil {
		_, _ = fmt.Fprintf(os.Stderr, "Indexing failed: %v\n", err)
		return err
	}

	tot := time.Now().Sub(start)
	_, _ = fmt.Fprintf(ctl.out, "Elapsed time:          %s\n", tot)

	return nil
}