Add labels from subscriptions.

Add colors
Add timestamp(what is timestamp when transaction was mined).
Add allowed filters for from query.

backend/moonstream/api.py

@@ -42,4 +42,4 @@ async def version_handler() -> data.VersionResponse:
 app.mount("/subscriptions", subscriptions_api)
 app.mount("/users", users_api)
 app.mount("/streams", streams_api)
-app.mount("/txinfo", txinfo_api)
+app.mount("/txinfo", txinfo_api)

backend/moonstream/data.py

@@ -4,6 +4,8 @@ Pydantic schemas for the Moonstream HTTP API
 from enum import Enum
 from typing import List, Optional
 
+from sqlalchemy.sql.operators import notendswith_op
+
 
 from pydantic import BaseModel, Field
 
@@ -109,15 +111,19 @@ class EthereumTransaction(BaseModel):
 
 
 class EthereumTransactionItem(BaseModel):
-    from_label: Optional[str] = "from_label"
-    to_label: Optional[str] = "to_label"
+    color: Optional[str]
+    from_label: Optional[str] = None
+    to_label: Optional[str] = None
     gas: int
     gasPrice: int
     value: int
+    nonce: Optional[str]
     from_address: Optional[str]  # = Field(alias="from")
     to_address: Optional[str]  # = Field(default=None, alias="to")
     hash: Optional[str] = None
     input: Optional[str] = None
+    timestamp: Optional[int] = None
+    subscription_type_id: Optional[str] = None
 
 
 class EthereumTransactionResponse(BaseModel):

backend/moonstream/routes/streams.py

@@ -3,6 +3,9 @@ The Moonstream subscriptions HTTP API
 """
 import logging
 from typing import Any, cast, Dict, List, Optional, Set, Union
+from pydantic.utils import to_camel
+
+from sqlalchemy.engine.base import Transaction
 
 from bugout.data import BugoutResource, BugoutResources
 from bugout.exceptions import BugoutResponseException
@@ -72,60 +75,119 @@ async def search_transactions(
 
     # get user subscriptions
 
+    token = request.state.token
+    params = {"user_id": str(request.state.user.id)}
+    try:
+        user_subscriptions_resources: BugoutResources = bc.list_resources(
+            token=token, params=params
+        )
+    except BugoutResponseException as e:
+        if e.detail == "Resources not found":
+            return data.EthereumTransactionResponse(stream=[])
+        raise HTTPException(status_code=e.status_code, detail=e.detail)
+    except Exception as e:
+        raise HTTPException(status_code=500)
+
+    subscriptions_addresses = [
+        resource.resource_data["address"]
+        for resource in user_subscriptions_resources.resources
+    ]
+
     if q == "" or q == " ":
-        token = request.state.token
-        params = {"user_id": str(request.state.user.id)}
-        try:
-            user_subscriptions_resources: BugoutResources = bc.list_resources(
-                token=token, params=params
-            )
-            print(user_subscriptions_resources)
-        except BugoutResponseException as e:
-            if e.detail == "Resources not found":
-                return data.EthereumTransactionResponse(stream=[])
-            raise HTTPException(status_code=e.status_code, detail=e.detail)
-        except Exception as e:
-            raise HTTPException(status_code=500)
-        user_subscriptions_resources
-        # search_query = search.normalized_search_query(q, filters, strict_filter_mode=False)
 
         filters = [
             or_(
-                EthereumTransaction.to_address == resource.resource_data["address"],
-                EthereumTransaction.from_address == resource.resource_data["address"],
+                EthereumTransaction.to_address == address,
+                EthereumTransaction.from_address == address,
             )
-            for resource in user_subscriptions_resources.resources
+            for address in subscriptions_addresses
         ]
         filters = or_(*filters)
 
     else:
-        print(f"query:|{q}|")
-        filters = database_search_query(q)
+        filters = database_search_query(q, allowed_addresses=subscriptions_addresses)
         if not filters:
             return data.EthereumTransactionResponse(stream=[])
         filters = and_(*filters)
 
-    transactions = db_session.query(EthereumTransaction).filter(filters).limit(25)
+    address_to_subscriptions = {
+        resource.resource_data["address"]: resource.resource_data
+        for resource in user_subscriptions_resources.resources
+    }
 
-    print(transactions)
-
-    response = [
-        data.EthereumTransactionItem(
-            gas=transaction.gas,
-            gasPrice=transaction.gas_price,
-            value=transaction.value,
-            from_address=transaction.from_address,
-            to_address=transaction.to_address,
-            hash=transaction.hash,
-            input=transaction.input,
+    ethereum_transactions = (
+        db_session.query(
+            EthereumTransaction.hash,
+            EthereumTransaction.block_number,
+            EthereumTransaction.from_address,
+            EthereumTransaction.to_address,
+            EthereumTransaction.gas,
+            EthereumTransaction.gas_price,
+            EthereumTransaction.input,
+            EthereumTransaction.nonce,
+            EthereumTransaction.value,
+            EthereumBlock.timestamp,
+        )
+        .join(EthereumBlock)
+        .filter(filters)
+        .limit(25)
+    )
+
+    response = []
+    for (
+        hash,
+        block_number,
+        from_address,
+        to_address,
+        gas,
+        gas_price,
+        input,
+        nonce,
+        value,
+        timestamp,
+    ) in ethereum_transactions:
+
+        subscription_type_id = None
+        from_label = None
+        to_label = None
+        color = None
+
+        if from_address in subscriptions_addresses:
+            from_label = address_to_subscriptions[from_address]["label"]
+            subscription_type_id = address_to_subscriptions[from_address][
+                "subscription_type_id"
+            ]
+            color = address_to_subscriptions[from_address]["color"]
+
+        if to_address in subscriptions_addresses:
+            subscription_type_id = address_to_subscriptions[to_address][
+                "subscription_type_id"
+            ]
+            to_label = address_to_subscriptions[to_address]["label"]
+            color = address_to_subscriptions[to_address]["color"]
+
+        response.append(
+            data.EthereumTransactionItem(
+                color=color,
+                from_label=from_label,
+                to_label=to_label,
+                gas=gas,
+                gasPrice=gas_price,
+                value=value,
+                from_address=from_address,
+                to_address=to_address,
+                hash=hash,
+                input=input,
+                nonce=nonce,
+                timestamp=timestamp,
+                subscription_type_id="1",
+            )
         )
-        for transaction in transactions
-    ]
 
     return data.EthereumTransactionResponse(stream=response)
 
 
-def database_search_query(q: str):
+def database_search_query(q: str, allowed_addresses: List[str]):
 
     filters = q.split("+")
     constructed_filters = []
@@ -146,6 +208,8 @@ def database_search_query(q: str):
             constructed_filters.append(EthereumTransaction.to_address == filter_value)
 
         if filter_type == "from" and filter_value:
+            if filter_value not in allowed_addresses:
+                continue
             constructed_filters.append(EthereumTransaction.from_address == filter_value)
 
         if filter_type == "address" and filter_value:

backend/moonstream/settings.py

@@ -14,13 +14,9 @@ MOONSTREAM_DATA_JOURNAL_ID = os.environ.get("MOONSTREAM_DATA_JOURNAL_ID")
 if MOONSTREAM_DATA_JOURNAL_ID is None:
     raise ValueError("MOONSTREAM_DATA_JOURNAL_ID environment variable must be set")
 
-MOONSTREAM_ADMIN_ACCESS_TOKEN = os.environ.get(
-    "MOONSTREAM_ADMIN_ACCESS_TOKEN"
-)
+MOONSTREAM_ADMIN_ACCESS_TOKEN = os.environ.get("MOONSTREAM_ADMIN_ACCESS_TOKEN", "")
 if MOONSTREAM_ADMIN_ACCESS_TOKEN is None:
-    raise ValueError(
-        "MOONSTREAM_ADMIN_ACCESS_TOKEN environment variable must be set"
-    )
+    raise ValueError("MOONSTREAM_ADMIN_ACCESS_TOKEN environment variable must be set")
 
 # Origin
 RAW_ORIGINS = os.environ.get("MOONSTREAM_CORS_ALLOWED_ORIGINS")

frontend/src/components/StreamEntry.js

@@ -42,6 +42,7 @@ const StreamEntry = ({ entry, filterCallback, filterConstants }) => {
   };
 
   const [showFullView] = useMediaQuery(["(min-width: 420px)"]);
+  console.log(entry);
 
   return (
     <Flex
@@ -321,9 +322,9 @@ const StreamEntry = ({ entry, filterCallback, filterConstants }) => {
                   flexGrow={1}
                 >
                   <Text mx={0} py="2px" fontSize="sm" w="calc(100%)" h="100%">
-                    {moment(entry.timestamp)
-                      // .format("DD MMM, YYYY, h:mm:ss")
-                      .fromNow()}{" "}
+                    {moment(entry.timestamp * 1000).format(
+                      "DD MMM, YYYY, HH:mm:ss"
+                    )}{" "}
                   </Text>
                 </Stack>
               )}