<commit_before><?php
/**
 * All action events are submitted via this producer
 */
namespace Minds\Core\EventStreams\Topics;

use Minds\Common\Urn;
use Minds\Core\EventStreams\ActionEvent;
use Minds\Core\EventStreams\EventInterface;
use Minds\Entities\User;
use Minds\Entities\Entity;
use Minds\Helpers\MagicAttributes;
use Pulsar\MessageBuilder;
use Pulsar\ProducerConfiguration;
use Pulsar\ConsumerConfiguration;
use Pulsar\Consumer;
use Pulsar\SchemaType;
use Pulsar\Result;

class ActionEventsTopic extends AbstractTopic implements TopicInterface
{
    /**
     * Sends action events to our stream
     * @param EventInterface $event
     */
    public function send(EventInterface $event): bool
    {
        if (!$event instanceof ActionEvent) {
            return false;
        }

        $tenant = $this->getPulsarTenant();
        $namespace = $this->getPulsarNamespace();
        $topic = 'event-action-' . $event->getAction();

        // Build the config and include the schema

        $config = new ProducerConfiguration();
        $config->setSchema(SchemaType::AVRO, "action", $this->getSchema(), []);

        $producer = $this->client()->createProducer("persistent://$tenant/$namespace/$topic", $config);

        // Build the message

        $builder = new MessageBuilder();
        $message = $builder
            //->setPartitionKey(0)
            ->setEventTimestamp($event->getTimestamp() ?: time())
            ->setContent(json_encode([
                'action' => $event->getAction(),
                'action_data' => $event->getActionData(),
                'user_guid' => (string) $event->getUser()->getGuid(),
                'entity_urn' => (string) $event->getEntity()->getUrn(),
                'entity_guid' => (string) $event->getEntity()->getGuid(),
                'entity_owner_guid' => (string) $event->getEntity()->getOwnerGuid(),
                'entity_type' => MagicAttributes::getterExists($event->getEntity(), 'getType') ? (string) $event->getEntity()->getType() : '',
                'entity_subtype' => MagicAttributes::getterExists($event->getEntity(), 'getSubtype') ? (string) $event->getEntity()->getSubtype() : '',
            ]))
            ->build();

        // Send the event to the stream

        $result = $producer->send($message);

        if ($result != Result::ResultOk) {
            return false;
        }

        return true;
    }

    /**
     * Consume stream events. Use a new $subscriptionId per service
     * eg. notifications, analytics, recomendations
     * @param string $subscriptionId
     * @param callable $callback - the logic for the event
     * @param string $topicRegex - defaults to * (all topics will be returned)
     * @return void
     */
    public function consume(string $subscriptionId, callable $callback, string $topicRegex = '*'): void
    {
        $tenant = $this->getPulsarTenant();
        $namespace = $this->getPulsarNamespace();
        $topicRegex = 'event-action-' . $topicRegex;
        //$topicRegex = '.*';

        $config = new ConsumerConfiguration();
        $config->setConsumerType(Consumer::ConsumerShared);
        $config->setSchema(SchemaType::AVRO, "action", $this->getSchema(), []);

        $consumer = $this->client()->subscribeWithRegex("persistent://$tenant/$namespace/$topicRegex", $subscriptionId, $config);

        while (true) {
            try {
                $message = $consumer->receive();
                $data = json_decode($message->getDataAsString(), true);

                /** @var User */
                $user = $this->entitiesBuilder->single($data['user_guid']);

                // If no user, something went wrong, but still skip
                if (!$user) {
                    $consumer->acknowledge($message);
                    continue;
                }<commit_msg>(fix): invalid entities or conflicts causing notifications to stall<commit_after><?php
/**
 * All action events are submitted via this producer
 */
namespace Minds\Core\EventStreams\Topics;

use Minds\Common\Urn;
use Minds\Core\EventStreams\ActionEvent;
use Minds\Core\EventStreams\EventInterface;
use Minds\Entities\User;
use Minds\Entities\Entity;
use Minds\Helpers\MagicAttributes;
use Pulsar\MessageBuilder;
use Pulsar\ProducerConfiguration;
use Pulsar\ConsumerConfiguration;
use Pulsar\Consumer;
use Pulsar\SchemaType;
use Pulsar\Result;

class ActionEventsTopic extends AbstractTopic implements TopicInterface
{
    /**
     * Sends action events to our stream
     * @param EventInterface $event
     */
    public function send(EventInterface $event): bool
    {
        if (!$event instanceof ActionEvent) {
            return false;
        }

        $tenant = $this->getPulsarTenant();
        $namespace = $this->getPulsarNamespace();
        $topic = 'event-action-' . $event->getAction();

        // Build the config and include the schema

        $config = new ProducerConfiguration();
        $config->setSchema(SchemaType::AVRO, "action", $this->getSchema(), []);

        $producer = $this->client()->createProducer("persistent://$tenant/$namespace/$topic", $config);

        // Build the message

        $builder = new MessageBuilder();
        $message = $builder
            //->setPartitionKey(0)
            ->setEventTimestamp($event->getTimestamp() ?: time())
            ->setContent(json_encode([
                'action' => $event->getAction(),
                'action_data' => $event->getActionData(),
                'user_guid' => (string) $event->getUser()->getGuid(),
                'entity_urn' => (string) $event->getEntity()->getUrn(),
                'entity_guid' => (string) $event->getEntity()->getGuid(),
                'entity_owner_guid' => (string) $event->getEntity()->getOwnerGuid(),
                'entity_type' => MagicAttributes::getterExists($event->getEntity(), 'getType') ? (string) $event->getEntity()->getType() : '',
                'entity_subtype' => MagicAttributes::getterExists($event->getEntity(), 'getSubtype') ? (string) $event->getEntity()->getSubtype() : '',
            ]))
            ->build();

        // Send the event to the stream

        $result = $producer->send($message);

        if ($result != Result::ResultOk) {
            return false;
        }

        return true;
    }

    /**
     * Consume stream events. Use a new $subscriptionId per service
     * eg. notifications, analytics, recomendations
     * @param string $subscriptionId
     * @param callable $callback - the logic for the event
     * @param string $topicRegex - defaults to * (all topics will be returned)
     * @return void
     */
    public function consume(string $subscriptionId, callable $callback, string $topicRegex = '*'): void
    {
        $tenant = $this->getPulsarTenant();
        $namespace = $this->getPulsarNamespace();
        $topicRegex = 'event-action-' . $topicRegex;
        //$topicRegex = '.*';

        $config = new ConsumerConfiguration();
        $config->setConsumerType(Consumer::ConsumerShared);
        $config->setSchema(SchemaType::AVRO, "action", $this->getSchema(), []);

        $consumer = $this->client()->subscribeWithRegex("persistent://$tenant/$namespace/$topicRegex", $subscriptionId, $config);

        while (true) {
            try {
                $message = $consumer->receive();
                $data = json_decode($message->getDataAsString(), true);

                /** @var User */
                $user = $this->entitiesBuilder->single($data['user_guid']);

                // If no user, something went wrong, but still skip
                if (!$user || !$user instanceof User) {
                    $consumer->acknowledge($message);
                    continue;
                }
{'fraction_non_alphanumeric': 0.14535335231685217, 'fraction_numerical': 0.0002588661661920787, 'mean_word_length': 2.529922338967565, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 101, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3945733', 'n_tokens_mistral': 2116, 'n_tokens_neox': 2056, 'n_words': 465}