feat(worker/mastodon): make mastodon worker revision and add federate…

…d api modification (#574)

Co-authored-by: brucexc <[email protected]>

internal/database/client.go

@@ -30,6 +30,7 @@ type Client interface {
 	SaveActivities(ctx context.Context, activities []*activityx.Activity) error
 	FindActivity(ctx context.Context, query model.ActivityQuery) (*activityx.Activity, *int, error)
 	FindActivities(ctx context.Context, query model.ActivitiesQuery) ([]*activityx.Activity, error)
+	FindFederatedActivities(ctx context.Context, query model.FederatedActivitiesQuery) ([]*activityx.Activity, error)
 	DeleteExpiredActivities(ctx context.Context, network network.Network, timestamp time.Time) error
 }
 

internal/database/dialer/postgres/client.go

@@ -293,6 +293,14 @@ func (c *client) FindActivities(ctx context.Context, query model.ActivitiesQuery
 	return nil, fmt.Errorf("not implemented")
 }
 
+func (c *client) FindFederatedActivities(ctx context.Context, query model.FederatedActivitiesQuery) ([]*activityx.Activity, error) {
+	if c.partition {
+		return c.findFederatedActivitiesPartitioned(ctx, query)
+	}
+
+	return nil, fmt.Errorf("not implemented")
+}
+
 // DeleteExpiredActivities deletes expired activities.
 func (c *client) DeleteExpiredActivities(ctx context.Context, network networkx.Network, timestamp time.Time) error {
 	if c.partition {

internal/database/dialer/postgres/client_partitioned.go

@@ -252,6 +252,75 @@ func (c *client) findActivitiesPartitioned(ctx context.Context, query model.Acti
 	return result, nil
 }
 
+// findActivitiesPartitioned finds activities.
+func (c *client) findFederatedActivitiesPartitioned(ctx context.Context, query model.FederatedActivitiesQuery) ([]*activityx.Activity, error) {
+	indexes, err := c.findFederatedIndexesPartitioned(ctx, query)
+	if err != nil {
+		return nil, fmt.Errorf("find indexes: %w", err)
+	}
+
+	partition := lo.GroupBy(indexes, func(query *table.Index) string {
+		activity := table.Activity{
+			ID:        query.ID,
+			Network:   query.Network,
+			Timestamp: query.Timestamp,
+		}
+
+		return activity.PartitionName(nil)
+	})
+
+	var (
+		result = make([]*activityx.Activity, 0)
+		locker sync.Mutex
+	)
+
+	errorGroup, errorCtx := errgroup.WithContext(ctx)
+
+	for tableName, index := range partition {
+		tableName, index := tableName, index
+
+		ids := lo.Map(index, func(index *table.Index, _ int) string {
+			return index.ID
+		})
+
+		errorGroup.Go(func() error {
+			tableActivities := make(table.Activities, 0)
+
+			if err := c.database.WithContext(errorCtx).Table(tableName).Where("id IN ?", lo.Uniq(ids)).Find(&tableActivities).Error; err != nil {
+				zap.L().Error("failed to find activities", zap.Error(err), zap.String("tableName", tableName))
+
+				return err
+			}
+
+			locker.Lock()
+			defer locker.Unlock()
+
+			activities, err := tableActivities.Export(index)
+			if err != nil {
+				return err
+			}
+
+			result = append(result, activities...)
+
+			return nil
+		})
+	}
+
+	if err := errorGroup.Wait(); err != nil {
+		return nil, err
+	}
+
+	lo.ForEach(result, func(activity *activityx.Activity, i int) {
+		result[i].Actions = lo.Slice(activity.Actions, 0, query.ActionLimit)
+	})
+
+	sort.SliceStable(result, func(i, j int) bool {
+		return result[i].Timestamp > result[j].Timestamp
+	})
+
+	return result, nil
+}
+
 // saveIndexesPartitioned saves indexes in partitioned tables.
 func (c *client) saveIndexesPartitioned(ctx context.Context, activities []*activityx.Activity) error {
 	indexes := make(table.Indexes, 0)
@@ -513,6 +582,109 @@ func (c *client) findIndexesPartitioned(ctx context.Context, query model.Activit
 	}
 }
 
+//nolint:gocognit
+func (c *client) findFederatedIndexesPartitioned(ctx context.Context, query model.FederatedActivitiesQuery) ([]*table.Index, error) {
+	index := table.Index{
+		Timestamp: time.Now(),
+	}
+
+	if query.EndTimestamp != nil && *query.EndTimestamp > 0 && *query.EndTimestamp < uint64(index.Timestamp.Unix()) {
+		index.Timestamp = time.Unix(int64(lo.FromPtr(query.EndTimestamp)), 0)
+	}
+
+	if query.Cursor != nil && query.Cursor.Timestamp < uint64(index.Timestamp.Unix()) {
+		index.Timestamp = time.Unix(int64(query.Cursor.Timestamp), 0)
+	}
+
+	partitionedNames := c.findIndexesPartitionTables(ctx, index)
+
+	if len(partitionedNames) == 0 {
+		return nil, nil
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	errorGroup, errorContext := errgroup.WithContext(ctx)
+	indexes := make([][]*table.Index, len(partitionedNames))
+	resultChan := make(chan int, len(partitionedNames))
+	errorChan := make(chan error)
+	stopChan := make(chan struct{})
+
+	var mutex sync.RWMutex
+
+	for partitionedIndex, partitionedName := range partitionedNames {
+		partitionedIndex, partitionedName := partitionedIndex, partitionedName
+
+		errorGroup.Go(func() error {
+			var result []*table.Index
+
+			databaseStatement := c.buildFederatedFindIndexesStatement(errorContext, partitionedName, query)
+
+			if err := databaseStatement.Find(&result).Error; err != nil {
+				return fmt.Errorf("failed to find indexes: %w", err)
+			}
+
+			mutex.Lock()
+			indexes[partitionedIndex] = result
+			mutex.Unlock()
+
+			select {
+			case <-stopChan:
+				return nil
+			case resultChan <- partitionedIndex:
+				return nil
+			}
+		})
+	}
+
+	go func() {
+		defer close(errorChan)
+
+		errorChan <- errorGroup.Wait()
+	}()
+
+	defer func() {
+		cancel()
+	}()
+
+	for {
+		select {
+		case err := <-errorChan:
+			if err != nil {
+				return nil, fmt.Errorf("failed to wait result: %w", err)
+			}
+		case <-resultChan:
+			result := make([]*table.Index, 0, query.Limit)
+			flag := true
+
+			mutex.RLock()
+
+			for _, data := range indexes {
+				data := data
+
+				if data == nil {
+					flag = false
+					break
+				}
+
+				result = append(result, data...)
+
+				if len(result) >= query.Limit {
+					close(stopChan)
+					mutex.RUnlock()
+
+					return result[:query.Limit], nil
+				}
+			}
+
+			mutex.RUnlock()
+
+			if flag {
+				return result, nil
+			}
+		}
+	}
+}
+
 // deleteExpiredActivitiesPartitioned deletes expired activities.
 func (c *client) deleteExpiredActivitiesPartitioned(ctx context.Context, network network.Network, timestamp time.Time) error {
 	var (
@@ -697,6 +869,65 @@ func (c *client) buildFindIndexesStatement(ctx context.Context, partition string
 	return databaseStatement.Order("timestamp DESC, index DESC").Limit(query.Limit)
 }
 
+// buildFindIndexesStatement builds the query indexes statement.
+func (c *client) buildFederatedFindIndexesStatement(ctx context.Context, partition string, query model.FederatedActivitiesQuery) *gorm.DB {
+	databaseStatement := c.database.WithContext(ctx).Table(partition)
+
+	if query.Distinct != nil && lo.FromPtr(query.Distinct) {
+		databaseStatement = databaseStatement.Select("DISTINCT (id) id, timestamp, index, network")
+	}
+
+	if query.Owner != nil {
+		databaseStatement = databaseStatement.Where("owner = ?", query.Owner)
+	}
+
+	if len(query.Owners) > 0 {
+		databaseStatement = databaseStatement.Where("owner IN ?", query.Owners)
+	}
+
+	if query.Status != nil {
+		databaseStatement = databaseStatement.Where("status = ?", query.Status)
+	}
+
+	if query.Direction != nil {
+		databaseStatement = databaseStatement.Where("direction = ?", query.Direction)
+	}
+
+	if query.StartTimestamp != nil && *query.StartTimestamp > 0 {
+		databaseStatement = databaseStatement.Where("timestamp >= ?", time.Unix(int64(*query.StartTimestamp), 0))
+	}
+
+	if query.EndTimestamp != nil && *query.EndTimestamp > 0 {
+		databaseStatement = databaseStatement.Where("timestamp <= ?", time.Unix(int64(*query.EndTimestamp), 0))
+	}
+
+	if query.Platform != "" {
+		databaseStatement = databaseStatement.Where("platform = ?", query.Platform)
+	}
+
+	if len(query.Platforms) > 0 {
+		databaseStatement = databaseStatement.Where("platform IN ?", query.Platforms)
+	}
+
+	if len(query.Tags) > 0 {
+		databaseStatement = databaseStatement.Where("tag IN ?", query.Tags)
+	}
+
+	if len(query.Types) > 0 {
+		databaseStatement = databaseStatement.Where("type IN ?", query.Types)
+	}
+
+	if len(query.Network) > 0 {
+		databaseStatement = databaseStatement.Where("network IN ?", query.Network)
+	}
+
+	if query.Cursor != nil && query.Cursor.Timestamp > 0 {
+		databaseStatement = databaseStatement.Where("timestamp < ? OR (timestamp = ? AND index < ?)", time.Unix(int64(query.Cursor.Timestamp), 0), time.Unix(int64(query.Cursor.Timestamp), 0), query.Cursor.Index)
+	}
+
+	return databaseStatement.Order("timestamp DESC, index DESC").Limit(query.Limit)
+}
+
 // buildActivitiesTableNames builds the activities table names.
 func (c *client) buildActivitiesTableNames(network network.Network, timestamp time.Time) string {
 	return fmt.Sprintf("%s_%s_%d_q%d", (*table.Activity).TableName(nil), network, timestamp.Year(), int(timestamp.Month()-1)/3+1)

internal/database/model/activity.go

@@ -2,6 +2,7 @@ package model
 
 import (
 	"github.com/rss3-network/node/schema/worker/decentralized"
+	"github.com/rss3-network/node/schema/worker/federated"
 	"github.com/rss3-network/protocol-go/schema"
 	activityx "github.com/rss3-network/protocol-go/schema/activity"
 	"github.com/rss3-network/protocol-go/schema/network"
@@ -34,3 +35,22 @@ type ActivitiesQuery struct {
 	Limit          int
 	ActionLimit    int
 }
+
+type FederatedActivitiesQuery struct {
+	Owner          *string
+	Cursor         *activityx.Activity
+	Status         *bool
+	Direction      *activityx.Direction
+	StartTimestamp *uint64
+	EndTimestamp   *uint64
+	Platform       string
+	Owners         []string
+	Network        []network.Network
+	Tags           []tag.Tag
+	Types          []schema.Type
+	Platforms      []federated.Platform
+	Distinct       *bool
+	RelatedActions *bool
+	Limit          int
+	ActionLimit    int
+}