plug-sdk/pkg/plug/algorithm.go

package plug

import (
	"context"
	"git.anthrove.art/Anthrove/otter-space-sdk/v4/pkg/database"
	"git.anthrove.art/Anthrove/otter-space-sdk/v4/pkg/models"
	log "github.com/sirupsen/logrus"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/trace"
	"slices"
	"time"
)

type User struct {
	UserFavoriteCount int64
	UserName          string
	UserID            string
}

type Favorites struct {
	Posts    []models.Post
	NextPage string
	LastPage string
}

type Plug interface {
	// GetFavoritePage
	// The API Key can be an empty string if it's not supplied by the user, in this case the default API Key should be used
	GetFavoritePage(ctx context.Context, apiKey string, userSource models.UserSource, pageIdentifier string) (Favorites, error)

	// GetUserProfile
	// The API Key can be an empty string if it's not supplied by the user, in this case the default API Key should be used
	GetUserProfile(ctx context.Context, apiKey string, userSource models.UserSource) (User, error)
}

func algorithm(ctx context.Context, plugInterface Plug, userSource models.UserSource, anthroveUserFavCount int64, deepScrape bool, apiKey string) (TaskSummery, error) {
	ctx, span := tracer.Start(ctx, "mainScrapeAlgorithm")
	defer span.End()

	span.SetAttributes(
		attribute.String("user_source_id", string(userSource.ID)),
		attribute.String("user_source_user_id", string(userSource.UserID)),
		attribute.String("user_source_source_id", string(userSource.SourceID)),
	)

	basicLoggingInfo := log.Fields{
		"user_source_id":        userSource.ID,
		"user_source_user_id":   userSource.UserID,
		"user_source_source_id": userSource.SourceID,
	}

	log.WithContext(ctx).WithFields(basicLoggingInfo).Info("Starting mainScrapeAlgorithm")

	taskSummery := TaskSummery{
		AddedPosts:   0,
		DeletedPosts: 0,
	}

	profile, err := plugInterface.GetUserProfile(ctx, apiKey, userSource)
	if err != nil {
		return taskSummery, err
	}

	userSource.UserID = models.UserID(profile.UserID)
	userSource.AccountUsername = profile.UserName

	nextPage := ""
	var newPosts []models.Post
	var anthroveFaves []models.UserFavorite

outer:
	for {
		if anthroveUserFavCount >= profile.UserFavoriteCount && profile.UserFavoriteCount > 0 {
			break outer
		}

		select {
		case <-ctx.Done():
			break outer
		default:
			span.AddEvent("Executing getFavorites request")
			favorites, err := plugInterface.GetFavoritePage(ctx, apiKey, userSource, nextPage)
			span.AddEvent("Finished executing getFavorites request")
			if err != nil {
				span.RecordError(err)
				span.SetStatus(codes.Error, err.Error())
				log.WithContext(ctx).WithFields(basicLoggingInfo).WithError(err).Error("Failed to execute favorites page")
				return taskSummery, err
			}

			if len(favorites.Posts) == 0 {
				span.AddEvent("No more favorites found")
				log.WithContext(ctx).WithFields(basicLoggingInfo).Info("No more favorites found")
				break outer
			}

			span.AddEvent("Executing BatchPostProcessingWithSummery")
			pageNewPosts, pageAnthroveFaves, err := BatchPostProcessingWithSummery(ctx, userSource, favorites.Posts)

			anthroveFaves = append(anthroveFaves, pageAnthroveFaves...)
			newPosts = append(newPosts, pageNewPosts...)

			span.AddEvent("Finished executing BatchPostProcessingWithSummery")
			if err != nil {
				span.RecordError(err)
				span.SetStatus(codes.Error, err.Error())
				log.WithContext(ctx).WithFields(basicLoggingInfo).WithError(err).Error("Failed in BatchPostProcessing")
				return taskSummery, err
			}

			if len(pageAnthroveFaves) <= 0 || len(favorites.Posts) != len(pageAnthroveFaves) {
				span.AddEvent("No more favorites found to add")
				log.WithContext(ctx).WithFields(basicLoggingInfo).Info("No more favorites found")
				break outer
			}

			nextPage = favorites.NextPage
			taskSummery.AddedPosts += len(pageAnthroveFaves)
		}
	}

	if len(newPosts) > 0 {
		span.AddEvent("Executing CreatePostInBatch")
		err = database.CreatePostInBatch(ctx, newPosts, BatchSize)
		if err != nil {
			span.RecordError(err)
			span.SetStatus(codes.Error, err.Error())
			log.WithContext(ctx).WithError(err).Error("Failed to create new posts in batch")
			return taskSummery, err
		}
		span.AddEvent("Created new posts in batch", trace.WithAttributes(attribute.Int("batch_size", BatchSize)))
		log.WithContext(ctx).WithFields(BasicLoggingFields).Info("Created new posts in batch")
	}

	if len(anthroveFaves) > 0 {
		span.AddEvent("Executing CreateUserFavoriteInBatch")

		for i, fav := range anthroveFaves {
			fav.CreatedAt = time.Now().Add(time.Millisecond * time.Duration(i) * -1)
		}

		slices.Reverse(anthroveFaves)

		err = database.CreateUserFavoriteInBatch(ctx, anthroveFaves, BatchSize)
		if err != nil {
			span.RecordError(err)
			span.SetStatus(codes.Error, err.Error())
			log.WithContext(ctx).WithError(err).WithFields(BasicLoggingFields).Error("Failed to create user favorites in batch")
			return taskSummery, err
		}
		span.AddEvent("Created user favorites in batch", trace.WithAttributes(attribute.Int("batch_size", BatchSize)))
		log.WithContext(ctx).WithFields(BasicLoggingFields).Info("Created user favorites in batch")
	}

	span.AddEvent("Completed scraping algorithm")
	log.WithContext(ctx).WithFields(basicLoggingInfo).Info("Completed scraping algorithm")
	return taskSummery, nil
}