[PAY-3881][PAY-3883] Add indexing/endpoints for separated collectible…

…s data (#11287)

.changeset/fresh-poems-visit.md

@@ -0,0 +1,5 @@
+---
+"@audius/sdk": minor
+---
+
+Add support for fetching collectibles

packages/discovery-provider/ddl/migrations/0118_add_collectibles.sql

@@ -0,0 +1,31 @@
+begin;
+
+create table if not exists collectibles (
+    user_id INTEGER NOT NULL,
+    data JSONB NOT NULL,
+    blockhash TEXT NOT NULL,
+    blocknumber INTEGER NOT NULL,
+    created_at TIMESTAMP WITH TIME ZONE DEFAULT now(),
+    updated_at TIMESTAMP WITH TIME ZONE DEFAULT now(),
+    CONSTRAINT pk_user_id PRIMARY KEY (user_id),
+    FOREIGN KEY (blocknumber) REFERENCES blocks(number) ON DELETE CASCADE
+);
+
+COMMENT ON TABLE collectibles IS 'Stores collectibles data for users';
+COMMENT ON COLUMN collectibles.user_id IS 'User ID of the person who owns the collectibles';
+COMMENT ON COLUMN collectibles.data IS 'Data about the collectibles';
+COMMENT ON COLUMN collectibles.blockhash IS 'Blockhash of the most recent block that changed the collectibles data';
+COMMENT ON COLUMN collectibles.blocknumber IS 'Block number of the most recent block that changed the collectibles data';
+
+INSERT INTO collectibles (user_id, data, blockhash, blocknumber)
+SELECT
+    u.user_id,
+    cid.data->'collectibles' AS data,
+    u.blockhash,
+    u.blocknumber
+FROM users u
+LEFT JOIN cid_data cid ON u.metadata_multihash = cid.cid
+WHERE u.has_collectibles = TRUE
+ON CONFLICT (user_id) DO NOTHING;
+
+commit;

packages/discovery-provider/integration_tests/tasks/entity_manager/test_user_entity_manager.py

@@ -14,6 +14,7 @@
 from src.challenges.challenge_event import ChallengeEvent
 from src.models.indexing.cid_data import CIDData
 from src.models.users.associated_wallet import AssociatedWallet
+from src.models.users.collectibles import Collectibles
 from src.models.users.user import User
 from src.queries.get_balances import IMMEDIATE_REFRESH_REDIS_PREFIX
 from src.solana.solana_client_manager import SolanaClientManager
@@ -1951,3 +1952,300 @@ def get_events_side_effect(_, tx_receipt):
                 mock.call(IMMEDIATE_REFRESH_REDIS_PREFIX, 1),
             ]
         )
+
+
+def test_add_user_collectibles(app, mocker):
+    """Tests adding user collectibles data"""
+    bus_mock = set_patches(mocker)
+
+    # setup db and mocked txs
+    with app.app_context():
+        db = get_db()
+        web3 = Web3()
+        update_task = UpdateTask(web3, bus_mock)
+
+    # Test data for valid collectibles
+    valid_collectibles = {
+        "collectibles": {
+            "order": ["collection1"],
+            "collection1": {},
+        }
+    }
+
+    # Test data for invalid collectibles (not a dict)
+    invalid_collectibles = {"collectibles": "not a dict"}
+
+    tx_receipts = {
+        "AddCollectiblesTx": [
+            {
+                "args": AttributeDict(
+                    {
+                        "_entityId": "",
+                        "_entityType": "Collectibles",
+                        "_userId": 1,
+                        "_action": "Create",
+                        "_metadata": json.dumps(
+                            {"cid": "", "data": valid_collectibles}
+                        ),
+                        "_signer": "user1wallet",
+                    }
+                )
+            },
+        ],
+        "InvalidCollectiblesTx": [
+            {
+                "args": AttributeDict(
+                    {
+                        "_entityId": "",
+                        "_entityType": "Collectibles",
+                        "_userId": 2,
+                        "_action": "Create",
+                        "_metadata": json.dumps(
+                            {"cid": "", "data": invalid_collectibles}
+                        ),
+                        "_signer": "user2wallet",
+                    }
+                )
+            },
+        ],
+    }
+
+    entity_manager_txs = [
+        AttributeDict({"transactionHash": update_task.web3.to_bytes(text=tx_receipt)})
+        for tx_receipt in tx_receipts
+    ]
+
+    def get_events_side_effect(_, tx_receipt):
+        return tx_receipts[tx_receipt["transactionHash"].decode("utf-8")]
+
+    mocker.patch(
+        "src.tasks.entity_manager.entity_manager.get_entity_manager_events_tx",
+        side_effect=get_events_side_effect,
+        autospec=True,
+    )
+
+    # Create user
+    entities = {
+        "users": [
+            {
+                "user_id": 1,
+                "handle": "user-1",
+                "wallet": "user1wallet",
+            },
+            {
+                "user_id": 2,
+                "handle": "user-2",
+                "wallet": "user2wallet",
+            },
+        ],
+    }
+    populate_mock_db(db, entities)
+
+    with db.scoped_session() as session:
+        # Test adding new collectibles
+        total_changes, _ = entity_manager_update(
+            update_task,
+            session,
+            [entity_manager_txs[0]],  # AddCollectiblesTx
+            block_number=0,
+            block_timestamp=BLOCK_DATETIME.timestamp(),
+            block_hash=hex(0),
+        )
+
+        # Verify collectibles were added
+        collectibles_data = (
+            session.query(Collectibles).filter(Collectibles.user_id == 1).first()
+        )
+        assert collectibles_data is not None
+        assert collectibles_data.data == valid_collectibles["collectibles"]
+        assert total_changes == 1
+
+        # Ensure collectibles flag was set to True
+        user = session.query(User).filter(User.user_id == 1).first()
+        assert user.has_collectibles
+
+        # Test invalid collectibles data
+        total_changes, _ = entity_manager_update(
+            update_task,
+            session,
+            [entity_manager_txs[1]],  # InvalidCollectiblesTx
+            block_number=0,
+            block_timestamp=BLOCK_DATETIME.timestamp(),
+            block_hash=hex(0),
+        )
+
+        # Verify no collectibles were added
+        assert total_changes == 0
+        current_data = (
+            session.query(Collectibles).filter(Collectibles.user_id == 2).first()
+        )
+        assert current_data is None
+
+
+def test_update_user_collectibles(app, mocker):
+    """Tests updating user collectibles data"""
+    bus_mock = set_patches(mocker)
+
+    # setup db and mocked txs
+    with app.app_context():
+        db = get_db()
+        web3 = Web3()
+        update_task = UpdateTask(web3, bus_mock)
+
+    updated_collectibles = {
+        "collectibles": {
+            "order": ["collection1"],
+            "collection1": {},
+            "collection2": {},
+        }
+    }
+
+    tx_receipts = {
+        "UpdateCollectiblesTx": [
+            {
+                "args": AttributeDict(
+                    {
+                        "_entityId": "",
+                        "_entityType": "Collectibles",
+                        "_userId": 1,
+                        "_action": "Update",
+                        "_metadata": json.dumps(
+                            {"cid": "", "data": updated_collectibles}
+                        ),
+                        "_signer": "user1wallet",
+                    }
+                )
+            },
+        ],
+        "InvalidCollectiblesTx": [
+            {
+                "args": AttributeDict(
+                    {
+                        "_entityId": "",
+                        "_entityType": "Collectibles",
+                        "_userId": 1,
+                        "_action": "Update",
+                        "_metadata": json.dumps(
+                            {"cid": "", "data": {"collectibles": "not a dict"}}
+                        ),
+                        "_signer": "user1wallet",
+                    }
+                )
+            },
+        ],
+        "EmptyCollectiblesTx": [
+            {
+                "args": AttributeDict(
+                    {
+                        "_entityId": "",
+                        "_entityType": "Collectibles",
+                        "_userId": 2,
+                        "_action": "Update",
+                        "_metadata": json.dumps(
+                            {"cid": "", "data": {"collectibles": {}}}
+                        ),
+                        "_signer": "user2wallet",
+                    }
+                )
+            },
+        ],
+    }
+
+    entity_manager_txs = [
+        AttributeDict({"transactionHash": update_task.web3.to_bytes(text=tx_receipt)})
+        for tx_receipt in tx_receipts
+    ]
+
+    def get_events_side_effect(_, tx_receipt):
+        return tx_receipts[tx_receipt["transactionHash"].decode("utf-8")]
+
+    mocker.patch(
+        "src.tasks.entity_manager.entity_manager.get_entity_manager_events_tx",
+        side_effect=get_events_side_effect,
+        autospec=True,
+    )
+
+    # Create user
+    entities = {
+        "users": [
+            {
+                "user_id": 1,
+                "handle": "user-1",
+                "wallet": "user1wallet",
+                "has_collectibles": True,
+            },
+            {
+                "user_id": 2,
+                "handle": "user-2",
+                "wallet": "user2wallet",
+                "has_collectibles": True,
+            },
+        ],
+        "collectibles_data": [
+            {
+                "user_id": 1,
+                "data": {"order": ["collection1"], "collection1": {}},
+            },
+            {
+                "user_id": 2,
+                "data": {"order": ["collection1"], "collection1": {}},
+            },
+        ],
+    }
+    populate_mock_db(db, entities)
+
+    with db.scoped_session() as session:
+        # Test updating existing collectibles
+        total_changes, _ = entity_manager_update(
+            update_task,
+            session,
+            [entity_manager_txs[0]],  # UpdateCollectiblesTx
+            block_number=0,
+            block_timestamp=BLOCK_DATETIME.timestamp(),
+            block_hash=hex(0),
+        )
+
+        # Verify collectibles were updated
+        updated_data = (
+            session.query(Collectibles).filter(Collectibles.user_id == 1).first()
+        )
+        assert updated_data is not None
+        assert updated_data.data == updated_collectibles["collectibles"]
+        assert total_changes == 1
+
+        # Test invalid collectibles data
+        total_changes, _ = entity_manager_update(
+            update_task,
+            session,
+            [entity_manager_txs[1]],  # InvalidCollectiblesTx
+            block_number=0,
+            block_timestamp=BLOCK_DATETIME.timestamp(),
+            block_hash=hex(0),
+        )
+        assert total_changes == 0
+        current_data = (
+            session.query(Collectibles).filter(Collectibles.user_id == 1).first()
+        )
+        assert current_data is not None
+        assert current_data.data == updated_collectibles["collectibles"]
+
+        # Test empty collectibles data
+        total_changes, _ = entity_manager_update(
+            update_task,
+            session,
+            [entity_manager_txs[2]],  # EmptyCollectiblesTx
+            block_number=1,
+            block_timestamp=BLOCK_DATETIME.timestamp(),
+            block_hash=hex(1),
+        )
+        assert total_changes == 1
+
+        current_data = (
+            session.query(Collectibles).filter(Collectibles.user_id == 2).first()
+        )
+        assert current_data is not None
+        assert current_data.data == {}
+
+        # Verify collectibles flag was set to False
+        user = session.query(User).filter(User.user_id == 2).first()
+        assert not user.has_collectibles

packages/discovery-provider/integration_tests/utils.py

@@ -43,6 +43,7 @@
 from src.models.tracks.track_route import TrackRoute
 from src.models.users.aggregate_user import AggregateUser
 from src.models.users.associated_wallet import AssociatedWallet, WalletChain
+from src.models.users.collectibles import Collectibles
 from src.models.users.email import EmailAccess, EncryptedEmail
 from src.models.users.supporter_rank_up import SupporterRankUp
 from src.models.users.usdc_purchase import PurchaseAccessType, USDCPurchase
@@ -182,6 +183,7 @@ def populate_mock_db(db, entities, block_offset=None):
         user_payout_wallet_history = entities.get("user_payout_wallet_history", [])
         encrypted_emails = entities.get("encrypted_emails", [])
         email_access = entities.get("email_access", [])
+        collectibles = entities.get("collectibles", [])
 
         num_blocks = max(
             len(tracks),
@@ -203,6 +205,7 @@ def populate_mock_db(db, entities, block_offset=None):
             len(track_price_history),
             len(album_price_history),
             len(user_payout_wallet_history),
+            len(collectibles),
         )
         for i in range(block_offset, block_offset + num_blocks):
             max_block = session.query(Block).filter(Block.number == i).first()
@@ -918,5 +921,15 @@ def populate_mock_db(db, entities, block_offset=None):
                 updated_at=email_access_meta.get("updated_at", datetime.now()),
             )
             session.add(email_access)
+        for i, collectible_data in enumerate(collectibles):
+            collectible_data_record = Collectibles(
+                user_id=collectible_data.get("user_id", i),
+                data=collectible_data.get("data", {}),
+                blockhash=collectible_data.get("blockhash", str(i + block_offset)),
+                blocknumber=collectible_data.get("blocknumber", i + block_offset),
+                created_at=collectible_data.get("created_at", datetime.now()),
+                updated_at=collectible_data.get("updated_at", datetime.now()),
+            )
+            session.add(collectible_data_record)
 
         session.commit()