Core/Notification/CassandraRepository.php

+<?php
+/**
+ * Cassandra Repository.
+ */
+
+namespace Minds\Core\Notification;
+
+use Minds\Core\Di\Di;
+use Minds\Common\Repository\Response;
+use Cassandra\Bigint;
+use Cassandra\Timeuuid;
+use Cassandra\Timestamp;
+use Minds\Common\Urn;
+use Minds\Core\Data\Cassandra\Prepared;
+
+class CassandraRepository
+{
+    const NOTIFICATION_TTL = ((60 * 60) * 24) * 30; // 30 days
+
+    /** @var $cql */
+    private $cql;
+
+    public function __construct($cql = null, $urn = null)
+    {
+        $this->cql = $cql ?: Di::_()->get('Database\Cassandra\Cql');
+        $this->urn = $urn ?: new Urn;
+    }
+
+    /**
+     * Get a list of notifications.
+     */
+    public function getList($opts)
+    {
+        $opts = array_merge([
+            'to_guid' => null,
+            'type_group' => null,
+            'uuid' => null,
+            'offset' => null,
+            'limit' => 12,
+        ], $opts);
+
+        if (!$opts['to_guid']) {
+            throw new \Exception('to_guid must be provided');
+        }
+
+        $statement = "SELECT * FROM notifications
+            WHERE to_guid = ?";
+        $values = [
+            new Bigint($opts['to_guid']),
+        ];
+
+        if ($opts['uuid']) {
+            $statement .= " AND uuid = ?";
+            $values[] = new Timeuuid($opts['uuid']);
+        }
+
+        if ($opts['type_group']) {
+            $statement = "SELECT * FROM notifications_by_type_group 
+                WHERE to_guid = ? 
+                AND type_group = ?";
+            $values[] = $opts['type_group'];
+        }
+
+        $query = new Prepared\Custom();
+        $query->query($statement, $values);
+        $query->setOpts([
+            'page_size' => $opts['limit'],
+            'paging_state_token' => base64_decode($opts['offset']),
+        ]);
+
+        try {
+            $result = $this->cql->request($query);
+        } catch (\Exception $e) {
+            return false;
+        }
+
+        $response = new Response();
+        foreach ($result as $row) {
+            $notification = new Notification();
+            $notification->setUuid($row['uuid']->uuid() ?: null)
+                ->setToGuid($row['to_guid'] ? (int) $row['to_guid']->value(): null)
+                ->setFromGuid($row['from_guid'] ? (int) $row['from_guid']->value(): null)
+                ->setEntityGuid((string) $row['entity_guid']) // REMOVE ONCE FULLY ON CASSANDRA
+                ->setEntityUrn($row['entity_urn'])
+                ->setCreatedTimestamp($row['created_timestamp'] ? $row['created_timestamp']->time() : null)
+                ->setReadTimestamp($row['read_timestamp'] ? $row['read_timestamp']->time() : null)
+                ->setType($row['type'])
+                ->setData(json_decode($row['data'], true));
+            $response[] = $notification;
+        }
+        $response->setPagingToken(base64_encode($result->pagingStateToken()));
+
+        return $response;
+    }
+
+    /**
+     * Get a single notification.
+     * @param $urn
+     * @return Notification
+     */
+    public function get($urn)
+    {
+        list ($to_guid, $uuid) = explode('-', $this->urn->setUrn($urn)->getNss());
+        
+        $response = $this->getList([
+            'to_guid' => $to_guid,
+            'uuid' => $uuid,
+            'limit' => 1,
+        ]);
+
+        return $response[0];
+    }
+
+
+    /**
+     * Add a notification to the database.
+     *
+     * @param Notification $notification
+     *
+     * @return Notification|bool
+     */
+    public function add($notification)
+    {
+        if (!$notification->getUuid()) {
+            $notification->setUuid((new Timeuuid($notification->getUuid() ?? time() * 1000))->uuid());
+        }
+
+        $statement = 'INSERT INTO notifications (
+            to_guid,
+            uuid,
+            type,
+            type_group,
+            from_guid,
+            entity_guid,
+            entity_urn,
+            created_timestamp,
+            read_timestamp,
+            data
+            ) VALUES (?,?,?,?,?,?,?,?,?, ?)
+            USING TTL ?';
+
+        $values = [
+            new Bigint($notification->getToGuid()),
+            new Timeuuid($notification->getUuid()),
+            (string) $notification->getType(),
+            Manager::getGroupFromType($notification->getType()),
+            new Bigint($notification->getFromGuid()),
+            (string) $notification->getEntityGuid(), // REMOVE ONCE FULLY ON CASSANDRA
+            (string) $notification->getEntityUrn(),
+            new Timestamp($notification->getCreatedTimestamp() ?: time()),
+            $notification->getReadTimestamp() ? new Timestamp($notification->getReadTimestamp()) : null,
+            json_encode($notification->getData()),
+            static::NOTIFICATION_TTL,
+        ];
+
+        $query = new Prepared\Custom();
+        $query->query($statement, $values);
+
+        try {
+            $success = $this->cql->request($query);
+        } catch (\Exception $e) {
+            return false;
+        }
+
+        return $success;
+    }
+
+    // TODO
+    public function update($notification, $fields)
+    {
+    }
+
+    // TODO
+    public function delete($uuid)
+    {
+    }
+}

Core/Notification/Events.php

@@ -75,6 +75,7 @@ class Events
                 ->setToGuid($params['to'])
                 ->setFromGuid($from_user->getGuid())
                 ->setEntityGuid($entityGuid)
+                ->setEntityUrn($entity->getUrn())
                 ->setType($params['notification_view'])
                 ->setData($data);
             

Core/Notification/Manager.php

@@ -9,6 +9,7 @@ use Minds\Common\Repository\Response;
 use Minds\Core\Config;
 use Minds\Core\Di\Di;
 use Minds\Entities\User;
+use Minds\Core\Features\Manager as FeaturesManager;
 
 class Manager
 {
@@ -19,17 +20,26 @@ class Manager
     /** @var Repository $repository */
     private $repository;
 
-    /** @var LegacyRepository $legacyRepository */
-    private $legacyRepository;
+    /** @var CassandraRepository $cassandraRepository */
+    private $cassandraRepository;
+
+    /** @var FeaturesManager $features */
+    private $features;
 
     /** @var User $user */
     private $user;
 
-    public function __construct($config = null, $repository = null, $legacyRepository = null)
+    public function __construct(
+        $config = null,
+        $repository = null,
+        $cassandraRepository = null,
+        $features = null
+    )
     {
         $this->config = $config ?: Di::_()->get('Config');
         $this->repository = $repository ?: new Repository;
-        $this->legacyRepository = $legacyRepository ?: new LegacyRepository;
+        $this->cassandraRepository = $cassandraRepository ?: new CassandraRepository;
+        $this->features = $features ?: new FeaturesManager;
     }
 
     /**
@@ -45,12 +55,15 @@ class Manager
 
     /**
      * Return a single notification
-     * @param $uuid
+     * @param $urn
      * @return Notification
      */
-    public function getSingle($uuid)
+    public function getSingle($urn)
     {
-        return $this->repository->get($uuid);
+        if (strpos($urn, 'urn:') !== FALSE) {
+            return $this->cassandraRepository->get($urn);
+        }
+        return $this->repository->get($urn);
     }
 
     /**
@@ -68,6 +81,8 @@ class Manager
             'offset' => '',
         ], $opts);
 
+        $opts['type_group'] = $opts['type'];
+
         switch ($opts['type']) {
             case "tags":
                 $opts['types'] = [
@@ -123,6 +138,10 @@ class Manager
                 break;
         }
 
+        if ($this->features->has('cassandra-notifications')) {
+            return $this->cassandraRepository->getList($opts);
+        }
+
         return $this->repository->getList($opts);
     }
 
@@ -134,6 +153,7 @@ class Manager
     public function add($notification)
     {
         try {
+            $this->cassandraRepository->add($notification);
             $uuid = $this->repository->add($notification);
 
             return $uuid;
@@ -145,4 +165,53 @@ class Manager
         }
     }
 
+    /**
+     * @param $type
+     * @return string
+     */
+    public static function getGroupFromType($type)
+    {
+        switch ($type) {
+            case 'tag':
+                return 'tags';
+                break;
+            case 'friends':
+            case 'welcome_chat':
+            case 'welcome_discover':
+                return 'subscriptions';
+                break;
+            case 'group_invite':
+            case 'group_kick':
+            case 'group_activity':
+                return 'groups';
+                break;
+            case 'comment':
+                return 'comments';
+                break;
+            case 'like':
+            case 'downvote':
+                return 'votes';
+                break;
+            case 'remind':
+                return 'reminds';
+                break;
+            case 'boost_gift':
+            case 'boost_submitted':
+            case 'boost_submitted_p2p':
+            case 'boost_request':
+            case 'boost_rejected':
+            case 'boost_revoked':
+            case 'boost_accepted':
+            case 'boost_completed':
+            case 'boost_peer_request':
+            case 'boost_peer_accepted':
+            case 'boost_peer_rejected':
+            case 'welcome_points':
+            case 'welcome_boost':
+                return 'boosts';
+                break;
+        }
+        return 'unknown';
+    }
+
 }

Core/Notification/Notification.php

@@ -9,14 +9,16 @@ use Minds\Traits\MagicAttributes;
 /**
  * Class Notification
  * @package Minds\Core\Notification
- * @method string getUUID()
- * @method Notification setUUID(string $value)
+ * @method string getUuid()
+ * @method Notification setUuid(string $value)
  * @method string getToGuid()
  * @method Notification setToGuid(string $value)
  * @method string getFromGuid()
  * @method Notification setFromGuid(string $value)
  * @method string getEntityGuid()
  * @method Notification setEntityGuid(string $value)
+ * @method string getEntityUrn()
+ * @method Notification getEntityUrn(string $value)
  * @method string getType()
  * @method Notification setType(string $value)
  * @method array getData()
@@ -44,6 +46,9 @@ class Notification
     /** @param string $entityGuid */
     private $entityGuid;
 
+    /** @param string $entityUrn */
+    private $entityUrn;
+
     /** @param string $type */
     private $type;
 
@@ -59,6 +64,15 @@ class Notification
     /** @var int $readTimestamp */
     private $readTimestamp;
 
+    /**
+     * Return the UUID of the notification
+     * @return string
+     */
+    public function getUrn()
+    {
+        return "urn:notification:" . implode('-', [ $this->toGuid, $this->uuid ]);
+    }
+
     /**
      * Export
      * @return array
@@ -66,9 +80,13 @@ class Notification
     public function export()
     {
         return [
+            'uuid' => $this->getUuid(),
             'toGuid' => $this->getToGuid(),
+            'to_guid' => $this->getToGuid(),
             'fromGuid' => $this->getFromGuid(),
+            'from_guid' => $this->getFromGuid(),
             'entityGuid' => $this->getEntityGuid(),
+            'entity_urn' => $this->getEntityUrn(),
             'type' => $this->getType(),
             'data' => $this->getData(),
         ];

Core/Notification/Repository.php

@@ -151,6 +151,7 @@ class Repository
         }
 
         $query = 'INSERT INTO notifications (
+            uuid,
             to_guid,
             from_guid,
             entity_guid,
@@ -162,6 +163,7 @@ class Repository
         $values = [];
         foreach ($notifications as $notification) {
             $values = array_merge($values, [
+                $notification->getUuid(),
                 $notification->getToGuid(),
                 $notification->getFromGuid(),
                 $notification->getEntityGuid(),
@@ -171,7 +173,7 @@ class Repository
             ]);
         }
 
-        $query .= implode(',', array_fill(0, count($notifications), '(?,?,?,?,?,?)'));
+        $query .= implode(',', array_fill(0, count($notifications), '(?,?,?,?,?,?,?)'));
 
         $query .= ' RETURNING UUID';
 

Core/Provisioner/Provisioners/cassandra-provision.cql

@@ -905,26 +905,28 @@ CREATE TABLE minds.translations (
     AND speculative_retry = '99PERCENTILE';
 
 CREATE TABLE minds.notifications (
-    owner_guid varint,
-    guid varint,
-    data text,
+    to_guid bigint,
+    uuid timeuuid,
     type text,
-    PRIMARY KEY (owner_guid, guid)
-) WITH CLUSTERING ORDER BY (guid DESC)
-    AND bloom_filter_fp_chance = 0.01
-    AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
-    AND comment = ''
-    AND compaction = {'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy', 'max_threshold': '32', 'min_threshold': '4'}
-    AND compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
-    AND crc_check_chance = 1.0
-    AND dclocal_read_repair_chance = 0.1
-    AND default_time_to_live = 0
-    AND gc_grace_seconds = 864000
-    AND max_index_interval = 2048
-    AND memtable_flush_period_in_ms = 0
-    AND min_index_interval = 128
-    AND read_repair_chance = 0.0
-    AND speculative_retry = '99PERCENTILE';
+    type_group text,
+    from_guid bigint,
+    entity_guid text,
+    entity_urn text,
+    created_timestamp timestamp,
+    read_timestamp timestamp,
+    data text,
+    PRIMARY KEY (to_guid, uuid)
+)  WITH CLUSTERING ORDER BY (uuid DESC);
+
+CREATE MATERIALIZED VIEW minds.notifications_by_type_group AS
+    SELECT *
+    FROM minds.notifications
+    WHERE to_guid IS NOT NULL 
+    AND uuid IS NOT NULL
+    AND type_group IS NOT NULL
+    PRIMARY KEY (to_guid, type_group, uuid)
+    WITH CLUSTERING ORDER BY (type_group DESC, uuid DESC);
+
 
 CREATE TABLE minds.friendsof (
     key text,

Spec/Core/Notification/CassandraRepositorySpec.php

+<?php
+
+namespace Spec\Minds\Core\Notification;
+
+use Minds\Core\Notification\CassandraRepository;
+use Minds\Core\Data\Cassandra\Client;
+use Minds\Core\Notification\Notification;
+use PhpSpec\ObjectBehavior;
+use Prophecy\Argument;
+use Spec\Minds\Mocks\Cassandra\Rows;
+use Cassandra\Timeuuid;
+use Cassandra\Bigint;
+use Cassandra\Timestamp;
+
+class CassandraRepositorySpec extends ObjectBehavior
+{
+    /** @var Client */
+    private $cql;
+
+    function let(Client $cql)
+    {
+        $this->beConstructedWith($cql);
+        $this->cql = $cql;
+    }
+
+    function it_is_initializable()
+    {
+        $this->shouldHaveType(CassandraRepository::class);
+    }
+
+    function it_should_return_a_notification_from_uuid()
+    {
+        $this->cql->request(Argument::that(function($prepared) {
+            $values = $prepared->build()['values'];
+            return $values[0]->value() == 123
+                && $values[1]->uuid() == 'uuid';
+        }))
+            ->willReturn(new Rows([
+                [
+                    'uuid' => new Timeuuid(time()),
+                    'to_guid' => new Bigint(123),
+                    'from_guid' => new Bigint(456),
+                    'entity_guid' => '789',
+                    'entity_urn' => 'urn:entity:789',
+                    'created_timestamp' => new Timestamp(time()),
+                    'read_timestamp' => null,
+                    'type' => 'like',
+                    'data' => '',
+                ]
+            ], ''));
+        $notification = $this->get('urn:notification:123-uuid');
+
+        $notification->getToGuid()
+            ->shouldBe(123);
+        $notification->getFromGuid()
+            ->shouldBe(456);
+        $notification->getEntityGuid()
+            ->shouldBe('789');
+        $notification->getEntityUrn()
+            ->shouldBe('urn:entity:789');
+        $notification->getCreatedTimestamp()
+            ->shouldBe(time());
+        $notification->getType()
+            ->shouldBe('like');
+    }
+
+    function it_should_add_to_database(Notification $notification)
+    {
+        $this->cql->request(Argument::that(function($prepared) {
+            $values = $prepared->build()['values'];
+            return $values[0]->value() == 123;
+        }))
+            ->shouldBeCalled()
+            ->willReturn(true);
+
+        $notification->getToGuid()
+            ->willReturn(123);
+        $notification->getUuid()
+            ->willReturn(null);
+        $notification->setUuid(Argument::type('string'))
+            ->shouldBeCalled();
+        $notification->getFromGuid()
+            ->willReturn(456);
+        $notification->getType()
+            ->willReturn('boost');
+        $notification->getEntityGuid()
+            ->willReturn(789);
+        $notification->getEntityUrn()
+            ->willReturn('urn:entity:789');
+        $notification->getCreatedTimestamp()
+            ->willReturn(time());
+        $notification->getReadTimestamp()
+            ->willReturn(null);
+        $notification->getData()
+            ->willReturn(null);
+
+        $this->add($notification)
+            ->shouldReturn(true);
+    }
+
+    function it_should_load_notification_for_user()
+    {
+        $this->cql->request(Argument::that(function($prepared) {
+            $statement = preg_replace('/\s+/', ' ', $prepared->build()['string']);
+            $values = $prepared->build()['values'];
+            return strpos($statement, 'SELECT * FROM notifications WHERE to_guid', 0) !== FALSE
+                && $values[0]->value() == 123;
+        }))
+            ->willReturn(new Rows([
+                [
+                    'uuid' => new Timeuuid(time()),
+                    'to_guid' => new Bigint(123),
+                    'from_guid' => new Bigint(456),
+                    'entity_guid' => '789',
+                    'entity_urn' => 'urn:entity:789',
+                    'created_timestamp' => new Timestamp(time()),
+                    'read_timestamp' => null,
+                    'type' => 'like',
+                    'data' => '',
+                ]
+            ], ''));
+
+        $notifications = $this->getList([ 
+            'limit' => 24,
+            'to_guid' => 123,
+        ]);
+        $notifications->shouldHaveCount(1);
+    }
+
+    function it_should_load_notification_from_type_group()
+    {
+        $this->cql->request(Argument::that(function($prepared) {
+            $statement = $prepared->build()['string'];
+            $values = $prepared->build()['values'];
+            return strpos($statement, 'SELECT * FROM notifications_by_type_group', 0) !== FALSE
+                && $values[0]->value() == 123
+                && $values[1] == 'votes';
+        }))
+            ->willReturn(new Rows([
+                [
+                    'uuid' => new Timeuuid(time()),
+                    'to_guid' => new Bigint(123),
+                    'from_guid' => new Bigint(456),
+                    'entity_guid' => '789',
+                    'entity_urn' => 'urn:entity:789',
+                    'created_timestamp' => new Timestamp(time()),
+                    'read_timestamp' => null,
+                    'type' => 'like',
+                    'data' => '',
+                ]
+            ], ''));
+
+        $notifications = $this->getList([ 
+            'limit' => 24,
+            'to_guid' => 123,
+            'type_group' => 'votes',
+        ]);
+        $notifications->shouldHaveCount(1);
+    }
+}

Spec/Core/Notification/ManagerSpec.php

@@ -7,7 +7,10 @@ use Minds\Core\Notification\LegacyRepository;
 use Minds\Core\Notification\Manager;
 use Minds\Core\Notification\Notification;
 use Minds\Core\Notification\Repository;
+use Minds\Core\Notification\CassandraRepository;
+use Minds\Core\Features\Manager as FeaturesManager;
 use PhpSpec\ObjectBehavior;
+use Prophecy\Argument;
 
 class ManagerSpec extends ObjectBehavior
 {
@@ -17,16 +20,25 @@ class ManagerSpec extends ObjectBehavior
     /** @var Repository */
     private $repository;
 
-    /** @var LegacyRepository */
-    private $legacyRepository;
+    /** @var CassandraRepository */
+    private $cassandraRepository;
 
-    function let(Config $config, Repository $repository, LegacyRepository $legacyRepository)
+    /** @var FeaturesManager */
+    private $features;
+
+    function let(
+        Config $config,
+        Repository $repository,
+        CassandraRepository $cassandraRepository,
+        FeaturesManager $features
+    )
     {
         $this->config = $config;
         $this->repository = $repository;
-        $this->legacyRepository = $legacyRepository;
+        $this->cassandraRepository = $cassandraRepository;
+        $this->features = $features;
 
-        $this->beConstructedWith($config, $repository, $legacyRepository);
+        $this->beConstructedWith($config, $repository, $cassandraRepository, $features);
     }
 
     function it_is_initializable()
@@ -42,4 +54,39 @@ class ManagerSpec extends ObjectBehavior
 
         $this->getSingle('1234')->shouldReturn($notification);
     }
+
+    function it_should_get_from_cassandra_if_urn_provided(Notification $notification)
+    {
+        $this->cassandraRepository->get('urn:notification:1234')
+            ->shouldBeCalled()
+            ->willReturn($notification);
+
+        $this->getSingle('urn:notification:1234')->shouldReturn($notification);
+    }
+
+    function it_should_get_list_from_cassandra_if_feature_enabled(Notification $notification)
+    {
+        $this->features->has('cassandra-notifications')
+            ->willReturn(true);
+
+        $this->cassandraRepository->getList(Argument::that(function($opts) {
+            return $opts['limit'] === 6;
+        }))
+            ->shouldBeCalled()
+            ->willReturn([ $notification ]);
+
+        $response = $this->getList([ 'limit' => 6 ]);
+        $response[0]->shouldBe($notification);
+    }
+
+    function it_should_add_to_both_repositories(Notification $notification)
+    {
+        $this->repository->add($notification)
+            ->shouldBeCalled();
+
+        $this->cassandraRepository->add($notification)
+            ->shouldBeCalled();
+
+        $this->add($notification);
+    }
 }