hub changes for stream/repost with extensions

hub/common.py

@@ -764,7 +764,16 @@ def __init__(self, got_version, expected_version):
             "claim_type": {"type": "byte"},
             "censor_type": {"type": "byte"},
             "trending_score": {"type": "double"},
-            "release_time": {"type": "long"}
+            "release_time": {"type": "long"},
+            # top-level keys of extensions dict indexed as keywords
+            "extensions": {
+                "type": "keyword"
+            },
+            # full contents of extensions dict available as "runtime" fields
+            "extensions_obj": {
+                "type": "object",
+                "dynamic": "runtime"
+            },
         }
     }
 }
@@ -788,7 +797,7 @@ def __init__(self, got_version, expected_version):
     'reposted_claim_id', 'repost_count', 'sd_hash',
     'trending_score', 'tx_num',
     'channel_tx_id', 'channel_tx_position', 'channel_height',  'reposted_tx_id',
-    'reposted_tx_position', 'reposted_height',
+    'reposted_tx_position', 'reposted_height', 'extensions',
 }
 
 TEXT_FIELDS = {
@@ -807,7 +816,12 @@ def __init__(self, got_version, expected_version):
     'channel_tx_position', 'channel_height',
 }
 
-ALL_FIELDS = RANGE_FIELDS | TEXT_FIELDS | FIELDS
+OBJECT_FIELDS = {
+    'extensions',
+}
+
+ALL_FIELDS = (RANGE_FIELDS | TEXT_FIELDS | FIELDS |
+    OBJECT_FIELDS | { f+'_obj' for f in OBJECT_FIELDS })
 
 REPLACEMENTS = {
     'claim_name': 'normalized_name',
@@ -826,6 +840,7 @@ def __init__(self, got_version, expected_version):
 
 
 def expand_query(**kwargs):
+    #print(f'expand_query: >>> {kwargs}')
     if "amount_order" in kwargs:
         kwargs["limit"] = 1
         kwargs["order_by"] = "effective_amount"
@@ -848,7 +863,10 @@ def expand_query(**kwargs):
             value = list(filter(None, value))
         if value is None or isinstance(value, list) and len(value) == 0:
             continue
+        if key in OBJECT_FIELDS and not isinstance(value, dict):
+            continue
         key = REPLACEMENTS.get(key, key)
+        #print(f'expand_query: *** {key} = {value}')
         if key in FIELDS:
             partial_id = False
             if key == 'claim_type':
@@ -911,6 +929,25 @@ def expand_query(**kwargs):
                             ]}
                         }
                     )
+            elif key in OBJECT_FIELDS:
+                def flatten(field, d):
+                    if isinstance(d, dict) and len(d) > 0:
+                        for k, v in d.items():
+                            subfield = f'{field}.{k}' if field else k
+                            yield from flatten(subfield, v)
+                    elif isinstance(d, dict):
+                        # require <field> be present
+                        yield {"exists": {"field": field}}
+                    elif isinstance(d, list):
+                        # require <field> match all values <d>
+                        yield {"bool": {"must": [{"match": {field: {"query": e}}} for e in d]}}
+                    else:
+                        # require <field> match value <d>
+                        yield {"match": {field: {"query": d}}}
+                # query field <key> for list of top-level dictionary keys
+                query['must'].append({"terms": {key: list(value.keys())}})
+                # query field <key>_obj for nested properties
+                query['must'].extend(flatten(f'{key}_obj', value))
             elif many:
                 query['must'].append({"terms": {key: value}})
             else:
@@ -1022,6 +1059,7 @@ def expand_query(**kwargs):
                 "sort": query["sort"]
             }
         }
+    #print(f'expand_query: <<< {query}')
     return query
 
 

hub/elastic_sync/db.py

@@ -1,6 +1,6 @@
 from typing import Optional, Set, Dict, List
 from concurrent.futures.thread import ThreadPoolExecutor
-from hub.schema.claim import guess_stream_type
+from hub.schema.claim import guess_stream_type, Claim
 from hub.schema.result import Censor
 from hub.common import hash160, STREAM_TYPES, CLAIM_TYPES, LRUCache
 from hub.db import SecondaryDB
@@ -43,9 +43,11 @@ async def prepare_claim_metadata_batch(self, claims: Dict[bytes, ResolveResult],
         metadatas.update(await self.get_claim_metadatas(list(needed_txos)))
 
         for claim_hash, claim in claims.items():
+            assert isinstance(claim, ResolveResult)
             metadata = metadatas.get((claim.tx_hash, claim.position))
             if not metadata:
                 continue
+            assert isinstance(metadata, Claim)
             if not metadata.is_stream or not metadata.stream.has_fee:
                 fee_amount = 0
             else:
@@ -98,16 +100,24 @@ async def prepare_claim_metadata_batch(self, claims: Dict[bytes, ResolveResult],
                 if reposted_metadata.is_stream and \
                         (reposted_metadata.stream.video.duration or reposted_metadata.stream.audio.duration):
                     reposted_duration = reposted_metadata.stream.video.duration or reposted_metadata.stream.audio.duration
+
+            extensions = None
             if metadata.is_stream:
                 meta = metadata.stream
+                extensions = meta.extensions.to_dict()
             elif metadata.is_channel:
                 meta = metadata.channel
             elif metadata.is_collection:
                 meta = metadata.collection
             elif metadata.is_repost:
                 meta = metadata.repost
+                modified = meta.reference.apply(reposted_metadata)
+                modified = getattr(modified, modified.claim_type)
+                if hasattr(modified, 'extensions'):
+                    extensions = modified.extensions.to_dict()
             else:
                 continue
+
             claim_tags = [tag for tag in meta.tags]
             claim_languages = [lang.language or 'none' for lang in meta.languages] or ['none']
             tags = list(set(claim_tags).union(set(reposted_tags)))
@@ -181,6 +191,8 @@ async def prepare_claim_metadata_batch(self, claims: Dict[bytes, ResolveResult],
                 'channel_tx_id': None if not claim.channel_tx_hash else claim.channel_tx_hash[::-1].hex(),
                 'channel_tx_position': claim.channel_tx_position,
                 'channel_height': claim.channel_height,
+                'extensions': list(extensions.keys()) if extensions else None,
+                'extensions_obj': extensions,
             }
 
             if metadata.is_repost and reposted_duration is not None:

hub/elastic_sync/service.py

@@ -134,6 +134,8 @@ async def start_index(self) -> bool:
         index_version = await self.get_index_version()
 
         res = await self.sync_client.indices.create(self.index, INDEX_DEFAULT_SETTINGS, ignore=400)
+        if 'error' in res:
+            self.log.warning("es index create failed: %s", res)
         acked = res.get('acknowledged', False)
 
         if acked:
@@ -189,7 +191,6 @@ async def batched_update_filter(items: typing.Dict[bytes, bytes], channel: bool,
                         await self.sync_client.update_by_query(
                             self.index, body=self.update_filter_query(censor_type, only_channels(batch), True),
                             slices=4)
-                    await self.sync_client.indices.refresh(self.index)
 
         if filtered_streams:
             await batched_update_filter(filtered_streams, False, Censor.SEARCH)
@@ -199,6 +200,8 @@ async def batched_update_filter(items: typing.Dict[bytes, bytes], channel: bool,
             await batched_update_filter(blocked_streams, False, Censor.RESOLVE)
         if blocked_channels:
             await batched_update_filter(blocked_channels, True, Censor.RESOLVE)
+        if filtered_streams or filtered_channels or blocked_streams or blocked_channels:
+            await self.sync_client.indices.refresh(self.index)
 
     @staticmethod
     def _upsert_claim_query(index, claim):

hub/schema/Makefile

@@ -1,5 +1,6 @@
 build:
-	rm types/v2/* -rf
+	rm -rf types/v2/*
 	touch types/v2/__init__.py
 	cd types/v2/ && protoc --python_out=. -I ../../../../../types/v2/proto/ ../../../../../types/v2/proto/*.proto
-	sed -e 's/^import\ \(.*\)_pb2\ /from . import\ \1_pb2\ /g' -i types/v2/*.py
+	cd types/v2/ && cp ../../../../../types/jsonschema/* ./
+	sed -e 's/^import\ \(.*\)_pb2\ /from . import\ \1_pb2\ /g' -i.bak types/v2/*.py