Merge branch 'main' into xai-crawlers

engineapi/alembic/versions/6d07739cb13e_live_at_for_metatx.py

@@ -0,0 +1,28 @@
+"""Live at for metatx
+
+Revision ID: 6d07739cb13e
+Revises: 71e888082a6d
+Create Date: 2023-12-06 14:33:04.814144
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '6d07739cb13e'
+down_revision = '71e888082a6d'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('call_requests', sa.Column('live_at', sa.DateTime(timezone=True), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('call_requests', 'live_at')
+    # ### end Alembic commands ###

engineapi/alembic/versions/7191eb70e99e_tx_hash_for_call_requests.py

@@ -0,0 +1,30 @@
+"""Tx hash for call requests
+
+Revision ID: 7191eb70e99e
+Revises: 6d07739cb13e
+Create Date: 2023-10-04 11:23:12.516797
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '7191eb70e99e'
+down_revision = '6d07739cb13e'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('call_requests', sa.Column('tx_hash', sa.VARCHAR(length=256), nullable=True))
+    op.create_unique_constraint(op.f('uq_call_requests_tx_hash'), 'call_requests', ['tx_hash'])
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(op.f('uq_call_requests_tx_hash'), 'call_requests', type_='unique')
+    op.drop_column('call_requests', 'tx_hash')
+    # ### end Alembic commands ###

engineapi/alembic/versions/71e888082a6d_leaderboard_metadata.py

@@ -0,0 +1,50 @@
+"""leaderboard metadata
+
+Revision ID: 71e888082a6d
+Revises: cc80e886e153
+Create Date: 2023-11-15 13:21:16.108399
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = "71e888082a6d"
+down_revision = "cc80e886e153"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "leaderboards",
+        sa.Column(
+            "blockchain_ids",
+            sa.ARRAY(sa.Integer()),
+            nullable=False,
+            server_default="{}",
+        ),
+    )
+    op.add_column(
+        "leaderboards",
+        sa.Column(
+            "wallet_connect", sa.Boolean(), nullable=False, server_default="false"
+        ),
+    )
+    op.add_column(
+        "leaderboards",
+        sa.Column(
+            "columns_names", postgresql.JSONB(astext_type=sa.Text()), nullable=True
+        ),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("leaderboards", "columns_names")
+    op.drop_column("leaderboards", "wallet_connect")
+    op.drop_column("leaderboards", "blockchain_ids")
+    # ### end Alembic commands ###

engineapi/engineapi/actions.py

@@ -5,7 +5,13 @@ from typing import List, Any, Optional, Dict, Union, Tuple, cast
 import uuid
 import logging
 
-from bugout.data import BugoutResource, BugoutSearchResult
+from bugout.data import (
+    BugoutResource,
+    BugoutSearchResult,
+    ResourcePermissions,
+    HolderType,
+    BugoutResourceHolder,
+)
 from eth_typing import Address
 from hexbytes import HexBytes
 import requests  # type: ignore
@@ -16,7 +22,14 @@ from sqlalchemy.engine import Row
 from web3 import Web3
 from web3.types import ChecksumAddress
 
-from .data import Score, LeaderboardScore, LeaderboardConfigUpdate, LeaderboardConfig
+from .data import (
+    Score,
+    LeaderboardScore,
+    LeaderboardConfigUpdate,
+    LeaderboardConfig,
+    LeaderboardPosition,
+    ColumnsNames,
+)
 from .contracts import Dropper_interface, ERC20_interface, Terminus_interface
 from .models import (
     DropperClaimant,
@@ -96,6 +109,10 @@ class LeaderboardVersionNotFound(Exception):
     pass
 
 
+class LeaderboardAssignResourceError(Exception):
+    pass
+
+
 BATCH_SIGNATURE_PAGE_SIZE = 500
 
 logger = logging.getLogger(__name__)
@@ -1268,7 +1285,7 @@ def get_leaderboard_score(
 
 def get_leaderboard_positions(
     db_session: Session,
-    leaderboard_id,
+    leaderboard_id: uuid.UUID,
     limit: int,
     offset: int,
     version_number: Optional[int] = None,
@@ -1481,31 +1498,47 @@ def create_leaderboard(
     title: str,
     description: Optional[str],
     token: Optional[Union[uuid.UUID, str]] = None,
+    wallet_connect: bool = False,
+    blockchain_ids: List[int] = [],
+    columns_names: ColumnsNames = None,
 ) -> Leaderboard:
     """
     Create a leaderboard
     """
 
+    if columns_names is not None:
+        columns_names = columns_names.dict()
+
     if not token:
         token = uuid.UUID(MOONSTREAM_ADMIN_ACCESS_TOKEN)
     try:
-        leaderboard = Leaderboard(title=title, description=description)
+        # deduplicate and sort
+        blockchain_ids = sorted(list(set(blockchain_ids)))
+
+        leaderboard = Leaderboard(
+            title=title,
+            description=description,
+            wallet_connect=wallet_connect,
+            blockchain_ids=blockchain_ids,
+            columns_names=columns_names,
+        )
         db_session.add(leaderboard)
         db_session.commit()
 
+        user = None
+        if token is not None:
+            user = bc.get_user(token=token)
+
         resource = create_leaderboard_resource(
             leaderboard_id=str(leaderboard.id),
-            token=token,
+            user_id=str(user.id) if user is not None else None,
         )
-
         leaderboard.resource_id = resource.id
-
         db_session.commit()
     except Exception as e:
         db_session.rollback()
         logger.error(f"Error creating leaderboard: {e}")
         raise LeaderboardCreateError(f"Error creating leaderboard: {e}")
-
     return leaderboard
 
 
@@ -1548,6 +1581,9 @@ def update_leaderboard(
     leaderboard_id: uuid.UUID,
     title: Optional[str],
     description: Optional[str],
+    wallet_connect: Optional[bool],
+    blockchain_ids: Optional[List[int]],
+    columns_names: Optional[ColumnsNames],
 ) -> Leaderboard:
     """
     Update a leaderboard
@@ -1561,6 +1597,23 @@ def update_leaderboard(
         leaderboard.title = title
     if description is not None:
         leaderboard.description = description
+    if wallet_connect is not None:
+        leaderboard.wallet_connect = wallet_connect
+    if blockchain_ids is not None:
+        # deduplicate and sort
+        blockchain_ids = sorted(list(set(blockchain_ids)))
+        leaderboard.blockchain_ids = blockchain_ids
+
+    if columns_names is not None:
+        if leaderboard.columns_names is not None:
+            current_columns_names = ColumnsNames(**leaderboard.columns_names)
+
+            for key, value in columns_names.dict(exclude_none=True).items():
+                setattr(current_columns_names, key, value)
+        else:
+            current_columns_names = columns_names
+
+        leaderboard.columns_names = current_columns_names.dict()
 
     db_session.commit()
 
@@ -1659,38 +1712,62 @@ def add_scores(
 # leaderboard access actions
 
 
-def create_leaderboard_resource(
+def create_leaderboard_resource(leaderboard_id: str, user_id: Optional[str] = None):
-    leaderboard_id: str, token: Union[Optional[uuid.UUID], str] = None
-) -> BugoutResource:
     resource_data: Dict[str, Any] = {
         "type": LEADERBOARD_RESOURCE_TYPE,
         "leaderboard_id": leaderboard_id,
     }
 
-    if token is None:
-        token = MOONSTREAM_ADMIN_ACCESS_TOKEN
     try:
         resource = bc.create_resource(
-            token=token,
+            token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
             application_id=MOONSTREAM_APPLICATION_ID,
             resource_data=resource_data,
             timeout=10,
         )
     except Exception as e:
         raise LeaderboardCreateError(f"Error creating leaderboard resource: {e}")
+
+    if user_id is not None:
+        try:
+            bc.add_resource_holder_permissions(
+                token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
+                resource_id=resource.id,
+                holder_permissions=BugoutResourceHolder(
+                    holder_type=HolderType.user,
+                    holder_id=user_id,
+                    permissions=[
+                        ResourcePermissions.ADMIN,
+                        ResourcePermissions.READ,
+                        ResourcePermissions.UPDATE,
+                        ResourcePermissions.DELETE,
+                    ],
+                ),
+            )
+        except Exception as e:
+            raise LeaderboardCreateError(
+                f"Error adding resource holder permissions: {e}"
+            )
+
     return resource
 
 
 def assign_resource(
     db_session: Session,
     leaderboard_id: uuid.UUID,
-    user_token: Union[uuid.UUID, str],
+    user_token: Optional[Union[uuid.UUID, str]] = None,
     resource_id: Optional[uuid.UUID] = None,
 ):
     """
     Assign a resource handler to a leaderboard
     """
 
+    ### get user_name from token
+
+    user = None
+    if user_token is not None:
+        user = bc.get_user(token=user_token)
+
     leaderboard = (
         db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one()  # type: ignore
     )
@@ -1698,11 +1775,9 @@ def assign_resource(
     if resource_id is not None:
         leaderboard.resource_id = resource_id
     else:
-        # Create resource via admin token
-
         resource = create_leaderboard_resource(
             leaderboard_id=str(leaderboard_id),
-            token=user_token,
+            user_id=user.id if user is not None else None,
         )
 
         leaderboard.resource_id = resource.id

engineapi/engineapi/auth.py

@@ -19,23 +19,17 @@ import argparse
 import base64
 import json
 import time
-from typing import Any, cast, Dict
+from typing import Any, Dict, Optional, cast
 
+import eth_keys
 from eip712.messages import EIP712Message, _hash_eip191_message
 from eth_account import Account
 from eth_account._utils.signing import sign_message_hash
-import eth_keys
+from eth_typing import ChecksumAddress
 from hexbytes import HexBytes
 from web3 import Web3
 
 
-AUTH_PAYLOAD_NAME = "MoonstreamAuthorization"
-AUTH_VERSION = "1"
-
-# By default, authorizations will remain active for 24 hours.
-DEFAULT_INTERVAL = 60 * 60 * 24
-
-
 class MoonstreamAuthorizationVerificationError(Exception):
     """
     Raised when invalid signer is provided.
@@ -48,12 +42,48 @@ class MoonstreamAuthorizationExpired(Exception):
     """
 
 
-class MoonstreamAuthorization(EIP712Message):
+class MoonstreamAuthorizationStructureError(Exception):
-    _name_: "string"
+    """
-    _version_: "string"
+    Raised when signature has incorrect structure.
+    """
 
-    address: "address"
+
-    deadline: "uint256"
+class MoonstreamAuthorization(EIP712Message):
+    _name_: "string"  # type: ignore
+    _version_: "string"  # type: ignore
+
+    address: "address"  # type: ignore
+    deadline: "uint256"  # type: ignore
+
+
+class MetaTXAuthorization(EIP712Message):
+    _name_: "string"  # type: ignore
+    _version_: "string"  # type: ignore
+
+    caller: "address"  # type: ignore
+    expires_at: "uint256"  # type: ignore
+
+
+EIP712_AUTHORIZATION_TYPES = {
+    "MoonstreamAuthorization": {
+        "name": "MoonstreamAuthorization",
+        "version": "1",
+        "eip712_message_class": MoonstreamAuthorization,
+        "primary_types": [
+            {"name": "address", "type": "address"},
+            {"name": "deadline", "type": int},
+        ],
+    },
+    "MetaTXAuthorization": {
+        "name": "MetaTXAuthorization",
+        "version": "1",
+        "eip712_message_class": MetaTXAuthorization,
+        "primary_types": [
+            {"name": "caller", "type": "address"},
+            {"name": "expires_at", "type": int},
+        ],
+    },
+}
 
 
 def sign_message(message_hash_bytes: HexBytes, private_key: HexBytes) -> HexBytes:
@@ -64,52 +94,77 @@ def sign_message(message_hash_bytes: HexBytes, private_key: HexBytes) -> HexByte
     return signed_message_bytes
 
 
-def authorize(deadline: int, address: str, private_key: HexBytes) -> Dict[str, Any]:
+def authorize(
-    message = MoonstreamAuthorization(
+    authorization_type: Dict[str, Any],
-        _name_=AUTH_PAYLOAD_NAME,
+    primary_types: Dict[str, Any],
-        _version_=AUTH_VERSION,
+    private_key: HexBytes,
-        address=address,
+    signature_name_output: str,
-        deadline=deadline,
+) -> Dict[str, Any]:
-    )
+    # Initializing instance of EIP712Message class
+    attrs: Dict[str, Any] = {
+        "_name_": authorization_type["name"],
+        "_version_": authorization_type["version"],
+    }
+    attrs.update(primary_types)
+    message = authorization_type["eip712_message_class"](**attrs)
 
+    # Generating message hash and signature
     msg_hash_bytes = HexBytes(_hash_eip191_message(message.signable_message))
 
     signed_message = sign_message(msg_hash_bytes, private_key)
 
-    api_payload: Dict[str, Any] = {
+    api_payload: Dict[str, Any] = {signature_name_output: signed_message.hex()}
-        "address": address,
+    api_payload.update(primary_types)
-        "deadline": deadline,
-        "signed_message": signed_message.hex(),
-    }
 
     return api_payload
 
 
-def verify(authorization_payload: Dict[str, Any]) -> bool:
+def verify(
+    authorization_type: Dict[str, Any],
+    authorization_payload: Dict[str, Any],
+    signature_name_input: str,
+) -> bool:
     """
     Verifies provided signature signer by correct address.
+
+    **Important** Assume that not address field is timefield (live_at, expires_at, deadline, etc)
     """
+    # Initializing instance of EIP712Message class
+    attrs: Dict[str, Any] = {
+        "_name_": authorization_type["name"],
+        "_version_": authorization_type["version"],
+    }
+
     time_now = int(time.time())
     web3_client = Web3()
-    address = Web3.toChecksumAddress(cast(str, authorization_payload["address"]))
-    deadline = cast(int, authorization_payload["deadline"])
-    signature = cast(str, authorization_payload["signed_message"])
 
-    message = MoonstreamAuthorization(
+    address: Optional[ChecksumAddress] = None
-        _name_=AUTH_PAYLOAD_NAME,
+    time_field: Optional[int] = None
-        _version_=AUTH_VERSION,
-        address=address,
-        deadline=deadline,
-    )
 
+    for pt in authorization_type["primary_types"]:
+        pt_name = pt["name"]
+        pt_type = pt["type"]
+        if pt_type == "address":
+            address = Web3.toChecksumAddress(cast(str, authorization_payload[pt_name]))
+            attrs[pt_name] = address
+        else:
+            time_field = cast(pt_type, authorization_payload[pt_name])
+            attrs[pt_name] = time_field
+    if address is None or time_field is None:
+        raise MoonstreamAuthorizationStructureError(
+            "Field address or time_field could not be None"
+        )
+
+    message = authorization_type["eip712_message_class"](**attrs)
+    signature = cast(str, authorization_payload[signature_name_input])
     signer_address = web3_client.eth.account.recover_message(
         message.signable_message, signature=signature
     )
     if signer_address != address:
         raise MoonstreamAuthorizationVerificationError("Invalid signer")
 
-    if deadline < time_now:
+    if time_field < time_now:
-        raise MoonstreamAuthorizationExpired("Deadline exceeded")
+        raise MoonstreamAuthorizationExpired("Time field exceeded")
 
     return True
 
@@ -121,15 +176,37 @@ def decrypt_keystore(keystore_path: str, password: str) -> HexBytes:
 
 
 def handle_authorize(args: argparse.Namespace) -> None:
-    address, private_key = decrypt_keystore(args.signer, args.password)
+    if args.authorization_type not in EIP712_AUTHORIZATION_TYPES:
-    authorization = authorize(args.deadline, address, private_key)
+        raise Exception("Provided unsupported EIP712 Authorization type")
+
+    authorization_type = EIP712_AUTHORIZATION_TYPES[args.authorization_type]
+    primary_types = json.loads(args.primary_types)
+    for ptk in authorization_type["primary_types"]:
+        if ptk["name"] not in primary_types:
+            raise Exception(f"Lost primary type: {ptk}")
+
+    _, private_key = decrypt_keystore(args.signer, args.password)
+    authorization = authorize(
+        authorization_type=authorization_type,
+        primary_types=primary_types,
+        private_key=private_key,
+        signature_name_output=args.signature_name_output,
+    )
     print(json.dumps(authorization))
 
 
 def handle_verify(args: argparse.Namespace) -> None:
+    if args.authorization_type not in EIP712_AUTHORIZATION_TYPES:
+        raise Exception("Provided unsupported EIP712 Authorization type")
+
+    authorization_type = EIP712_AUTHORIZATION_TYPES[args.authorization_type]
     payload_json = base64.decodebytes(args.payload).decode("utf-8")
     payload = json.loads(payload_json)
-    verify(payload)
+    verify(
+        authorization_type=authorization_type,
+        authorization_payload=payload,
+        signature_name_input=args.signature_name_input,
+    )
     print("Verified!")
 
 
@@ -140,13 +217,6 @@ def generate_cli() -> argparse.ArgumentParser:
     subcommands = parser.add_subparsers()
 
     authorize_parser = subcommands.add_parser("authorize")
-    authorize_parser.add_argument(
-        "-t",
-        "--deadline",
-        type=int,
-        default=int(time.time()) + DEFAULT_INTERVAL,
-        help="Authorization deadline (seconds since epoch timestamp).",
-    )
     authorize_parser.add_argument(
         "-s",
         "--signer",
@@ -159,6 +229,30 @@ def generate_cli() -> argparse.ArgumentParser:
         required=False,
         help="(Optional) password for signing account. If you don't provide it here, you will be prompte for it.",
     )
+    authorize_parser.add_argument(
+        "-t",
+        "--authorization-type",
+        required=True,
+        choices=[k for k in EIP712_AUTHORIZATION_TYPES.keys()],
+        help="One of supported EIP712 Message authorization types",
+    )
+    authorize_parser.add_argument(
+        "--primary-types",
+        required=True,
+        help="Primary types for specified EIP712 Message authorization in JSON format {0}. Available keys: {1}".format(
+            {"name_1": "value", "name_2": "value"},
+            [
+                f"{v['primary_types']} for {k}"
+                for k, v in EIP712_AUTHORIZATION_TYPES.items()
+            ],
+        ),
+    )
+    authorize_parser.add_argument(
+        "--signature-name-output",
+        type=str,
+        default="signed_message",
+        help="Key in output dictionary of signature",
+    )
     authorize_parser.set_defaults(func=handle_authorize)
 
     verify_parser = subcommands.add_parser("verify")
@@ -168,6 +262,19 @@ def generate_cli() -> argparse.ArgumentParser:
         required=True,
         help="Base64-encoded payload to verify",
     )
+    verify_parser.add_argument(
+        "-t",
+        "--authorization-type",
+        required=True,
+        choices=[k for k in EIP712_AUTHORIZATION_TYPES.keys()],
+        help="One of supported EIP712 Message authorization types",
+    )
+    verify_parser.add_argument(
+        "--signature-name-input",
+        type=str,
+        default="signed_message",
+        help="Key for signature in payload",
+    )
     verify_parser.set_defaults(func=handle_verify)
 
     return parser

engineapi/engineapi/contracts_actions.py

@@ -2,10 +2,10 @@ import argparse
 import json
 import logging
 import uuid
-from datetime import timedelta
+from datetime import datetime, timedelta
 from typing import Any, Dict, List, Optional, Tuple
 
-from sqlalchemy import func, text
+from sqlalchemy import func, or_, text
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.engine import Row
 from sqlalchemy.exc import IntegrityError, NoResultFound
@@ -100,7 +100,9 @@ def parse_call_request_response(
         method=obj[0].method,
         request_id=str(obj[0].request_id),
         parameters=obj[0].parameters,
+        tx_hash=obj[0].tx_hash,
         expires_at=obj[0].expires_at,
+        live_at=obj[0].live_at,
         created_at=obj[0].created_at,
         updated_at=obj[0].updated_at,
     )
@@ -326,13 +328,14 @@ def delete_registered_contract(
     return (registered_contract, blockchain)
 
 
-def request_calls(
+def create_request_calls(
     db_session: Session,
     metatx_requester_id: uuid.UUID,
     registered_contract_id: Optional[uuid.UUID],
     contract_address: Optional[str],
     call_specs: List[data.CallSpecification],
     ttl_days: Optional[int] = None,
+    live_at: Optional[int] = None,
 ) -> int:
     """
     Batch creates call requests for the given registered contract.
@@ -350,6 +353,11 @@ def request_calls(
         if ttl_days <= 0:
             raise ValueError("ttl_days must be positive")
 
+    if live_at is not None:
+        assert live_at == int(live_at)
+        if live_at <= 0:
+            raise ValueError("live_at must be positive")
+
     # Check that the moonstream_user_id matches a RegisteredContract with the given id or address
     query = db_session.query(RegisteredContract).filter(
         RegisteredContract.metatx_requester_id == metatx_requester_id
@@ -406,6 +414,7 @@ def request_calls(
             request_id=specification.request_id,
             parameters=specification.parameters,
             expires_at=expires_at,
+            live_at=datetime.fromtimestamp(live_at) if live_at is not None else None,
         )
 
         db_session.add(request)
@@ -422,7 +431,7 @@ def request_calls(
     return len(call_specs)
 
 
-def get_call_requests(
+def get_call_request(
     db_session: Session,
     request_id: uuid.UUID,
 ) -> Tuple[CallRequest, RegisteredContract]:
@@ -472,9 +481,14 @@ def list_call_requests(
     limit: int = 10,
     offset: Optional[int] = None,
     show_expired: bool = False,
+    live_after: Optional[int] = None,
+    metatx_requester_id: Optional[uuid.UUID] = None,
 ) -> List[Row[Tuple[CallRequest, RegisteredContract, CallRequestType]]]:
     """
-    List call requests for the given moonstream_user_id
+    List call requests.
+
+    Argument moonstream_user_id took from authorization workflow. And if it is specified
+    then user has access to call_requests before live_at param.
     """
     if caller is None:
         raise ValueError("caller must be specified")
@@ -507,6 +521,23 @@ def list_call_requests(
             CallRequest.expires_at > func.now(),
         )
 
+    # If user id not specified, do not show call_requests before live_at.
+    # Otherwise check show_before_live_at argument from query parameter
+    if metatx_requester_id is not None:
+        query = query.filter(
+            CallRequest.metatx_requester_id == metatx_requester_id,
+        )
+    else:
+        query = query.filter(
+            or_(CallRequest.live_at < func.now(), CallRequest.live_at == None)
+        )
+
+    if live_after is not None:
+        assert live_after == int(live_after)
+        if live_after <= 0:
+            raise ValueError("live_after must be positive")
+        query = query.filter(CallRequest.live_at >= datetime.fromtimestamp(live_after))
+
     if offset is not None:
         query = query.offset(offset)
 
@@ -551,6 +582,46 @@ def delete_requests(
     return requests_to_delete_num
 
 
+def complete_call_request(
+    db_session: Session,
+    tx_hash: str,
+    call_request_id: uuid.UUID,
+    caller: str,
+) -> CallRequest:
+    results = (
+        db_session.query(CallRequest, RegisteredContract)
+        .join(
+            RegisteredContract,
+            CallRequest.registered_contract_id == RegisteredContract.id,
+        )
+        .filter(CallRequest.id == call_request_id)
+        .filter(CallRequest.caller == caller)
+        .all()
+    )
+
+    if len(results) == 0:
+        raise CallRequestNotFound("Call request with given ID not found")
+    elif len(results) != 1:
+        raise Exception(
+            f"Incorrect number of results found for request_id {call_request_id}"
+        )
+    call_request, registered_contract = results[0]
+
+    call_request.tx_hash = tx_hash
+
+    try:
+        db_session.add(call_request)
+        db_session.commit()
+    except Exception as err:
+        logger.error(
+            f"complete_call_request -- error updating in database: {repr(err)}"
+        )
+        db_session.rollback()
+        raise
+
+    return (call_request, registered_contract)
+
+
 def handle_register(args: argparse.Namespace) -> None:
     """
     Handles the register command.
@@ -633,7 +704,7 @@ def handle_request_calls(args: argparse.Namespace) -> None:
 
     try:
         with db.yield_db_session_ctx() as db_session:
-            request_calls(
+            create_request_calls(
                 db_session=db_session,
                 moonstream_user_id=args.moonstream_user_id,
                 registered_contract_id=args.registered_contract_id,

engineapi/engineapi/data.py

@@ -284,6 +284,7 @@ class CreateCallRequestsAPIRequest(BaseModel):
     contract_address: Optional[str] = None
     specifications: List[CallSpecification] = Field(default_factory=list)
     ttl_days: Optional[int] = None
+    live_at: Optional[int] = None
 
     # Solution found thanks to https://github.com/pydantic/pydantic/issues/506
     @root_validator
@@ -305,7 +306,9 @@ class CallRequestResponse(BaseModel):
     method: str
     request_id: str
     parameters: Dict[str, Any]
+    tx_hash: Optional[str] = None
     expires_at: Optional[datetime] = None
+    live_at: Optional[datetime] = None
     created_at: datetime
     updated_at: datetime
 
@@ -326,6 +329,10 @@ class CallRequestResponse(BaseModel):
         return Web3.toChecksumAddress(v)
 
 
+class CompleteCallRequestsAPIRequest(BaseModel):
+    tx_hash: str
+
+
 class QuartilesResponse(BaseModel):
     percentile_25: Dict[str, Any]
     percentile_50: Dict[str, Any]
@@ -365,11 +372,22 @@ class LeaderboardScore(BaseModel):
     points_data: Dict[str, Any]
 
 
+class ColumnsNames(BaseModel):
+    rank: Optional[str] = None
+    address: Optional[str] = None
+    score: Optional[str] = None
+    points_data: Optional[str] = None
+    points_data_fields: Optional[Dict[str, str]] = None
+
+
 class Leaderboard(BaseModel):
     id: UUID
     title: str
     description: Optional[str] = None
     resource_id: Optional[UUID] = None
+    wallet_connect: bool = False
+    blockchain_ids: List[int] = Field(default_factory=list)
+    columns_names: Optional[ColumnsNames] = None
     created_at: datetime
     updated_at: datetime
 
@@ -385,6 +403,9 @@ class LeaderboardInfoResponse(BaseModel):
 class LeaderboardCreateRequest(BaseModel):
     title: str
     description: Optional[str] = None
+    wallet_connect: bool = False
+    blockchain_ids: List[int] = Field(default_factory=list)
+    columns_names: Optional[ColumnsNames] = None
 
 
 class LeaderboardCreatedResponse(BaseModel):
@@ -392,6 +413,9 @@ class LeaderboardCreatedResponse(BaseModel):
     title: str
     description: Optional[str] = None
     resource_id: Optional[UUID] = None
+    wallet_connect: bool = False
+    blockchain_ids: List[int] = Field(default_factory=list)
+    columns_names: Optional[ColumnsNames] = None
     created_at: datetime
     updated_at: datetime
 
@@ -404,6 +428,9 @@ class LeaderboardUpdatedResponse(BaseModel):
     title: str
     description: Optional[str] = None
     resource_id: Optional[UUID] = None
+    wallet_connect: bool = False
+    blockchain_ids: List[int] = Field(default_factory=list)
+    columns_names: Optional[ColumnsNames] = None
     created_at: datetime
     updated_at: datetime
 
@@ -414,6 +441,9 @@ class LeaderboardUpdatedResponse(BaseModel):
 class LeaderboardUpdateRequest(BaseModel):
     title: Optional[str] = None
     description: Optional[str] = None
+    wallet_connect: bool = False
+    blockchain_ids: List[int] = Field(default_factory=list)
+    columns_names: Optional[ColumnsNames] = None
 
 
 class LeaderboardDeletedResponse(BaseModel):
@@ -421,6 +451,9 @@ class LeaderboardDeletedResponse(BaseModel):
     title: str
     description: Optional[str] = None
     resource_id: Optional[UUID] = None
+    wallet_connect: bool = False
+    blockchain_ids: List[int] = Field(default_factory=list)
+    columns_names: Optional[ColumnsNames] = None
     created_at: datetime
     updated_at: datetime
 

engineapi/engineapi/middleware.py

@@ -6,17 +6,24 @@ from uuid import UUID
 
 from bugout.data import BugoutResource, BugoutResources, BugoutUser
 from bugout.exceptions import BugoutResponseException
-from fastapi import HTTPException, Request, Response
+from eip712.messages import EIP712Message, _hash_eip191_message
+from eth_account.messages import encode_defunct
+from fastapi import Depends, Header, HTTPException, Request, Response
+from fastapi.security import OAuth2PasswordBearer
+from hexbytes import HexBytes
 from pydantic import AnyHttpUrl, parse_obj_as
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.middleware.cors import CORSMiddleware
 from starlette.responses import Response
 from starlette.types import ASGIApp
 from web3 import Web3
+from web3.auto import w3 as w3_auto
 
 from . import data
 from .auth import (
+    EIP712_AUTHORIZATION_TYPES,
     MoonstreamAuthorizationExpired,
+    MoonstreamAuthorizationStructureError,
     MoonstreamAuthorizationVerificationError,
     verify,
 )
@@ -26,13 +33,170 @@ from .settings import (
     BUGOUT_REQUEST_TIMEOUT_SECONDS,
     BUGOUT_RESOURCE_TYPE_APPLICATION_CONFIG,
     MOONSTREAM_ADMIN_ACCESS_TOKEN,
-    MOONSTREAM_APPLICATION_ID,
     MOONSTREAM_ADMIN_ID,
+    MOONSTREAM_APPLICATION_ID,
 )
 from .settings import bugout_client as bc
 
 logger = logging.getLogger(__name__)
 
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+
+class InvalidAuthHeaderFormat(Exception):
+    """
+    Raised when authorization header not pass validation.
+    """
+
+
+class BugoutUnverifiedAuth(Exception):
+    """
+    Raised when attempted access by unverified Brood account.
+    """
+
+
+class BugoutAuthWrongApp(Exception):
+    """
+    Raised when user does not belong to this application.
+    """
+
+
+def parse_auth_header(auth_header: str) -> Tuple[str, str]:
+    """
+    Returns: auth_format and user_token passed in authorization header.
+    """
+    auth_list = auth_header.split()
+    if len(auth_list) != 2:
+        raise InvalidAuthHeaderFormat("Wrong authorization header")
+
+    return auth_list[0], auth_list[1]
+
+
+def bugout_auth(token: str) -> BugoutUser:
+    """
+    Extended bugout.get_user with additional checks.
+    """
+    user: BugoutUser = bc.get_user(token)
+    if not user.verified:
+        raise BugoutUnverifiedAuth("Only verified accounts can have access")
+    if str(user.application_id) != str(MOONSTREAM_APPLICATION_ID):
+        raise BugoutAuthWrongApp("User does not belong to this application")
+
+    return user
+
+
+def brood_auth(token: UUID) -> BugoutUser:
+    try:
+        user: BugoutUser = bugout_auth(token=token)
+    except BugoutUnverifiedAuth:
+        logger.info(f"Attempted access by unverified Brood account: {user.id}")
+        raise EngineHTTPException(
+            status_code=403,
+            detail="Only verified accounts can have access",
+        )
+    except BugoutAuthWrongApp:
+        raise EngineHTTPException(
+            status_code=403,
+            detail="User does not belong to this application",
+        )
+    except BugoutResponseException as e:
+        raise EngineHTTPException(
+            status_code=e.status_code,
+            detail=e.detail,
+        )
+    except Exception as e:
+        logger.error(f"Error processing Brood response: {str(e)}")
+        raise EngineHTTPException(
+            status_code=500,
+            detail="Internal server error",
+        )
+
+    return user
+
+
+async def request_user_auth(
+    token: UUID = Depends(oauth2_scheme),
+) -> BugoutUser:
+    user = brood_auth(token=token)
+
+    return user
+
+
+async def request_none_or_user_auth(
+    authorization: str = Header(None),
+) -> Optional[BugoutUser]:
+    """
+    Fetch Bugout user if authorization token provided.
+    """
+    user: Optional[BugoutUser] = None
+    if authorization is not None:
+        token: str = ""
+        try:
+            _, token = parse_auth_header(auth_header=authorization)
+        except InvalidAuthHeaderFormat:
+            raise EngineHTTPException(
+                status_code=403, detail="Wrong authorization header"
+            )
+        except Exception as e:
+            logger.error(f"Error parsing auth header: {str(e)}")
+            raise EngineHTTPException(status_code=500, detail="Internal server error")
+
+        if token != "":
+            user = brood_auth(token=token)
+
+    return user
+
+
+async def metatx_verify_header(
+    authorization: str = Header(None),
+) -> Optional[Dict[str, Any]]:
+    message: Optional[Dict[str, Any]] = None
+    if authorization is not None:
+        try:
+            auth_format, user_token = parse_auth_header(auth_header=authorization)
+        except InvalidAuthHeaderFormat:
+            raise EngineHTTPException(
+                status_code=403, detail="Wrong authorization header"
+            )
+        except Exception as e:
+            logger.error(f"Error parsing auth header: {str(e)}")
+            raise EngineHTTPException(status_code=500, detail="Internal server error")
+
+        if auth_format != "metatx":
+            raise EngineHTTPException(
+                status_code=403,
+                detail=f"Wrong authorization header format: {auth_format}",
+            )
+
+        try:
+            json_payload_str = base64.b64decode(user_token).decode("utf-8")
+            payload = json.loads(json_payload_str)
+            verify(
+                authorization_type=EIP712_AUTHORIZATION_TYPES["MetaTXAuthorization"],
+                authorization_payload=payload,
+                signature_name_input="signature",
+            )
+            message = {
+                "caller": Web3.toChecksumAddress(payload.get("caller")),
+                "expires_at": payload.get("expires_at"),
+            }
+        except MoonstreamAuthorizationVerificationError as e:
+            logger.info("MetaTX authorization verification error: %s", e)
+            raise EngineHTTPException(status_code=403, detail="Invalid signer")
+        except MoonstreamAuthorizationExpired as e:
+            logger.info("MetaTX authorization expired: %s", e)
+            raise EngineHTTPException(status_code=403, detail="Authorization expired")
+        except MoonstreamAuthorizationStructureError as e:
+            logger.info("MetaTX authorization incorrect structure error: %s", e)
+            raise EngineHTTPException(
+                status_code=403, detail="Incorrect signature structure"
+            )
+        except Exception as e:
+            logger.error("Unexpected exception: %s", e)
+            raise EngineHTTPException(status_code=500, detail="Internal server error")
+
+        return message
+
 
 class BroodAuthMiddleware(BaseHTTPMiddleware):
     """
@@ -59,30 +223,33 @@ class BroodAuthMiddleware(BaseHTTPMiddleware):
         if path in self.whitelist.keys() and self.whitelist[path] == method:
             return await call_next(request)
 
-        authorization_header = request.headers.get("authorization")
+        authorization = request.headers.get("authorization")
-        if authorization_header is None:
+        if authorization is None:
             return Response(
-                status_code=403, content="No authorization header passed with request"
+                status_code=403,
+                content="No authorization header passed with request",
             )
-        user_token_list = authorization_header.split()
-        if len(user_token_list) != 2:
-            return Response(status_code=403, content="Wrong authorization header")
-        user_token: str = user_token_list[-1]
 
         try:
-            user: BugoutUser = bc.get_user(user_token)
+            _, user_token = parse_auth_header(auth_header=authorization)
-            if not user.verified:
+        except InvalidAuthHeaderFormat:
-                logger.info(
+            return Response(status_code=403, content="Wrong authorization header")
-                    f"Attempted journal access by unverified Brood account: {user.id}"
+        except Exception as e:
-                )
+            logger.error(f"Error parsing auth header: {str(e)}")
-                return Response(
+            return Response(status_code=500, content="Internal server error")
-                    status_code=403,
+
-                    content="Only verified accounts can access journals",
+        try:
-                )
+            user: BugoutUser = bugout_auth(token=user_token)
-            if str(user.application_id) != str(MOONSTREAM_APPLICATION_ID):
+        except BugoutUnverifiedAuth:
-                return Response(
+            logger.info(f"Attempted access by unverified Brood account: {user.id}")
-                    status_code=403, content="User does not belong to this application"
+            return Response(
-                )
+                status_code=403,
+                content="Only verified accounts can have access",
+            )
+        except BugoutAuthWrongApp:
+            return Response(
+                status_code=403, content="User does not belong to this application"
+            )
         except BugoutResponseException as e:
             return Response(status_code=e.status_code, content=e.detail)
         except Exception as e:
@@ -139,9 +306,15 @@ class EngineAuthMiddleware(BaseHTTPMiddleware):
                 authorization_header_components[-1]
             ).decode("utf-8")
 
-            json_payload = json.loads(json_payload_str)
+            payload = json.loads(json_payload_str)
-            verified = verify(json_payload)
+            verified = verify(
-            address = json_payload.get("address")
+                authorization_type=EIP712_AUTHORIZATION_TYPES[
+                    "MoonstreamAuthorization"
+                ],
+                authorization_payload=payload,
+                signature_name_input="signed_message",
+            )
+            address = payload.get("address")
             if address is not None:
                 address = Web3.toChecksumAddress(address)
             else:

engineapi/engineapi/models.py

@@ -1,6 +1,7 @@
 import uuid
 
 from sqlalchemy import (
+    ARRAY,
     DECIMAL,
     VARCHAR,
     BigInteger,
@@ -314,8 +315,10 @@ class CallRequest(Base):
     method = Column(String, nullable=False, index=True)
     request_id = Column(DECIMAL, nullable=False, index=True)
     parameters = Column(JSONB, nullable=False)
+    tx_hash = Column(VARCHAR(256), unique=True, nullable=True)
 
     expires_at = Column(DateTime(timezone=True), nullable=True, index=True)
+    live_at = Column(DateTime(timezone=True), nullable=True)
 
     created_at = Column(
         DateTime(timezone=True), server_default=utcnow(), nullable=False
@@ -335,7 +338,6 @@ class CallRequest(Base):
 
 class Leaderboard(Base):  # type: ignore
     __tablename__ = "leaderboards"
-    # __table_args__ = (UniqueConstraint("dropper_contract_id", "address"),)
 
     id = Column(
         UUID(as_uuid=True),
@@ -347,6 +349,10 @@ class Leaderboard(Base):  # type: ignore
     title = Column(VARCHAR(128), nullable=False)
     description = Column(String, nullable=True)
     resource_id = Column(UUID(as_uuid=True), nullable=True, index=True)
+    blockchain_ids = Column(ARRAY(Integer), nullable=False, default=[])
+
+    wallet_connect = Column(Boolean, default=False, nullable=False)
+    columns_names = Column(JSONB, nullable=True)
     created_at = Column(
         DateTime(timezone=True), server_default=utcnow(), nullable=False
     )

engineapi/engineapi/routes/leaderboard.py

@@ -2,7 +2,7 @@
 Leaderboard API.
 """
 import logging
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Any, Union
 from uuid import UUID
 
 from bugout.exceptions import BugoutResponseException
@@ -92,9 +92,12 @@ app.add_middleware(
     "",
     response_model=List[data.LeaderboardPosition],
     tags=["Public Endpoints"],
-    include_in_schema=False,
 )
-@app.get("/", response_model=List[data.LeaderboardPosition], tags=["Public Endpoints"])
+@app.get(
+    "/",
+    response_model=List[data.LeaderboardPosition],
+    tags=["Public Endpoints"],
+)
 async def leaderboard(
     leaderboard_id: UUID = Query(..., description="Leaderboard ID"),
     limit: int = Query(10),
@@ -108,7 +111,7 @@ async def leaderboard(
 
     ### Check if leaderboard exists
     try:
-        actions.get_leaderboard_by_id(db_session, leaderboard_id)
+        leaderboard = actions.get_leaderboard_by_id(db_session, leaderboard_id)
     except NoResultFound as e:
         raise EngineHTTPException(
             status_code=404,
@@ -119,8 +122,9 @@ async def leaderboard(
         raise EngineHTTPException(status_code=500, detail="Internal server error")
 
     leaderboard_positions = actions.get_leaderboard_positions(
-        db_session, leaderboard_id, limit, offset, version
+        db_session, leaderboard.id, limit, offset, version
     )
+
     result = [
         data.LeaderboardPosition(
             address=position.address,
@@ -150,7 +154,6 @@ async def create_leaderboard(
     Authorization: str = AuthHeader,
 ) -> data.LeaderboardCreatedResponse:
     """
-
     Create leaderboard.
     """
 
@@ -162,6 +165,9 @@ async def create_leaderboard(
             title=leaderboard.title,
             description=leaderboard.description,
             token=token,
+            wallet_connect=leaderboard.wallet_connect,
+            blockchain_ids=leaderboard.blockchain_ids,
+            columns_names=leaderboard.columns_names,
         )
     except actions.LeaderboardCreateError as e:
         logger.error(f"Error while creating leaderboard: {e}")
@@ -177,12 +183,15 @@ async def create_leaderboard(
     # Add resource to the leaderboard
 
     return data.LeaderboardCreatedResponse(
-        id=created_leaderboard.id,  # type: ignore
+        id=created_leaderboard.id,
-        title=created_leaderboard.title,  # type: ignore
+        title=created_leaderboard.title,
-        description=created_leaderboard.description,  # type: ignore
+        description=created_leaderboard.description,
-        resource_id=created_leaderboard.resource_id,  # type: ignore
+        resource_id=created_leaderboard.resource_id,
-        created_at=created_leaderboard.created_at,  # type: ignore
+        wallet_connect=created_leaderboard.wallet_connect,
-        updated_at=created_leaderboard.updated_at,  # type: ignore
+        blockchain_ids=created_leaderboard.blockchain_ids,
+        columns_names=created_leaderboard.columns_names,
+        created_at=created_leaderboard.created_at,
+        updated_at=created_leaderboard.updated_at,
     )
 
 
@@ -226,6 +235,9 @@ async def update_leaderboard(
             leaderboard_id=leaderboard_id,
             title=leaderboard.title,
             description=leaderboard.description,
+            wallet_connect=leaderboard.wallet_connect,
+            blockchain_ids=leaderboard.blockchain_ids,
+            columns_names=leaderboard.columns_names,
         )
     except actions.LeaderboardUpdateError as e:
         logger.error(f"Error while updating leaderboard: {e}")
@@ -239,12 +251,15 @@ async def update_leaderboard(
         raise EngineHTTPException(status_code=500, detail="Internal server error")
 
     return data.LeaderboardUpdatedResponse(
-        id=updated_leaderboard.id,  # type: ignore
+        id=updated_leaderboard.id,
-        title=updated_leaderboard.title,  # type: ignore
+        title=updated_leaderboard.title,
-        description=updated_leaderboard.description,  # type: ignore
+        description=updated_leaderboard.description,
-        resource_id=updated_leaderboard.resource_id,  # type: ignore
+        resource_id=updated_leaderboard.resource_id,
-        created_at=updated_leaderboard.created_at,  # type: ignore
+        wallet_connect=updated_leaderboard.wallet_connect,
-        updated_at=updated_leaderboard.updated_at,  # type: ignore
+        blockchain_ids=updated_leaderboard.blockchain_ids,
+        columns_names=updated_leaderboard.columns_names,
+        created_at=updated_leaderboard.created_at,
+        updated_at=updated_leaderboard.updated_at,
     )
 
 
@@ -299,11 +314,15 @@ async def delete_leaderboard(
         raise EngineHTTPException(status_code=500, detail="Internal server error")
 
     return data.LeaderboardDeletedResponse(
-        id=deleted_leaderboard.id,  # type: ignore
+        id=deleted_leaderboard.id,
-        title=deleted_leaderboard.title,  # type: ignore
+        title=deleted_leaderboard.title,
-        description=deleted_leaderboard.description,  # type: ignore
+        description=deleted_leaderboard.description,
-        created_at=deleted_leaderboard.created_at,  # type: ignore
+        resource_id=deleted_leaderboard.resource_id,
-        updated_at=deleted_leaderboard.updated_at,  # type: ignore
+        wallet_connect=deleted_leaderboard.wallet_connect,
+        blockchain_ids=deleted_leaderboard.blockchain_ids,
+        columns_names=deleted_leaderboard.columns_names,
+        created_at=deleted_leaderboard.created_at,
+        updated_at=deleted_leaderboard.updated_at,
     )
 
 
@@ -336,12 +355,15 @@ async def get_leaderboards(
 
     results = [
         data.Leaderboard(
-            id=leaderboard.id,  # type: ignore
+            id=leaderboard.id,
-            title=leaderboard.title,  # type: ignore
+            title=leaderboard.title,
-            description=leaderboard.description,  # type: ignore
+            description=leaderboard.description,
-            resource_id=leaderboard.resource_id,  # type: ignore
+            resource_id=leaderboard.resource_id,
-            created_at=leaderboard.created_at,  # type: ignore
+            wallet_connect=leaderboard.wallet_connect,
-            updated_at=leaderboard.updated_at,  # type: ignore
+            blockchain_ids=leaderboard.blockchain_ids,
+            columns_names=leaderboard.columns_names,
+            created_at=leaderboard.created_at,
+            updated_at=leaderboard.updated_at,
         )
         for leaderboard in leaderboards
     ]
@@ -453,7 +475,7 @@ async def quartiles(
     """
     ### Check if leaderboard exists
     try:
-        actions.get_leaderboard_by_id(db_session, leaderboard_id)
+        leaderboard = actions.get_leaderboard_by_id(db_session, leaderboard_id)
     except NoResultFound as e:
         raise EngineHTTPException(
             status_code=404,
@@ -472,12 +494,14 @@ async def quartiles(
         logger.error(f"Error while getting quartiles: {e}")
         raise EngineHTTPException(status_code=500, detail="Internal server error")
 
-    return data.QuartilesResponse(
+    result = data.QuartilesResponse(
-        percentile_25={"address": q1[0], "score": q1[1], "rank": q1[2]},
+        percentile_25={"address": q1.address, "rank": q1.rank, "score": q1.score},
-        percentile_50={"address": q2[0], "score": q2[1], "rank": q2[2]},
+        percentile_50={"address": q2.address, "rank": q2.rank, "score": q2.score},
-        percentile_75={"address": q3[0], "score": q3[1], "rank": q3[2]},
+        percentile_75={"address": q3.address, "rank": q3.rank, "score": q3.score},
     )
 
+    return result
+
 
 @app.get(
     "/position",
@@ -503,7 +527,7 @@ async def position(
 
     ### Check if leaderboard exists
     try:
-        actions.get_leaderboard_by_id(db_session, leaderboard_id)
+        leaderboard = actions.get_leaderboard_by_id(db_session, leaderboard_id)
     except NoResultFound as e:
         raise EngineHTTPException(
             status_code=404,
@@ -540,7 +564,9 @@ async def position(
 
 
 @app.get(
-    "/rank", response_model=List[data.LeaderboardPosition], tags=["Public Endpoints"]
+    "/rank",
+    response_model=List[data.LeaderboardPosition],
+    tags=["Public Endpoints"],
 )
 async def rank(
     leaderboard_id: UUID = Query(..., description="Leaderboard ID"),
@@ -556,7 +582,7 @@ async def rank(
 
     ### Check if leaderboard exists
     try:
-        actions.get_leaderboard_by_id(db_session, leaderboard_id)
+        leaderboard = actions.get_leaderboard_by_id(db_session, leaderboard_id)
     except NoResultFound as e:
         raise EngineHTTPException(
             status_code=404,
@@ -574,14 +600,15 @@ async def rank(
         offset=offset,
         version_number=version,
     )
+
     results = [
         data.LeaderboardPosition(
-            address=rank_position.address,
+            address=position.address,
-            score=rank_position.score,
+            score=position.score,
-            rank=rank_position.rank,
+            rank=position.rank,
-            points_data=rank_position.points_data,
+            points_data=position.points_data,
         )
-        for rank_position in leaderboard_rank
+        for position in leaderboard_rank
     ]
     return results
 

engineapi/engineapi/routes/metatx.py

@@ -9,12 +9,19 @@ import logging
 from typing import Dict, List, Optional
 from uuid import UUID
 
-from fastapi import Body, Depends, FastAPI, Path, Query, Request
+from bugout.data import BugoutUser
+from fastapi import Body, Depends, FastAPI, Form, Path, Query, Request
 from sqlalchemy.exc import NoResultFound
 from sqlalchemy.orm import Session
 
 from .. import contracts_actions, data, db
-from ..middleware import BroodAuthMiddleware, BugoutCORSMiddleware, EngineHTTPException
+from ..middleware import (
+    BugoutCORSMiddleware,
+    EngineHTTPException,
+    metatx_verify_header,
+    request_none_or_user_auth,
+    request_user_auth,
+)
 from ..settings import DOCS_TARGET_PATH
 from ..version import VERSION
 
@@ -34,15 +41,6 @@ tags_metadata = [
 ]
 
 
-whitelist_paths = {
-    "/metatx/openapi.json": "GET",
-    f"/metatx/{DOCS_TARGET_PATH}": "GET",
-    "/metatx/blockchains": "GET",
-    "/metatx/contracts/types": "GET",
-    "/metatx/requests/types": "GET",
-    "/metatx/requests": "GET",
-}
-
 app = FastAPI(
     title=TITLE,
     description=DESCRIPTION,
@@ -53,9 +51,6 @@ app = FastAPI(
     redoc_url=f"/{DOCS_TARGET_PATH}",
 )
 
-
-app.add_middleware(BroodAuthMiddleware, whitelist=whitelist_paths)
-
 app.add_middleware(
     BugoutCORSMiddleware,
     allow_credentials=True,
@@ -89,11 +84,11 @@ async def blockchains_route(
     response_model=List[data.RegisteredContractResponse],
 )
 async def list_registered_contracts_route(
-    request: Request,
     blockchain: Optional[str] = Query(None),
     address: Optional[str] = Query(None),
     limit: int = Query(10),
     offset: Optional[int] = Query(None),
+    user: BugoutUser = Depends(request_user_auth),
     db_session: Session = Depends(db.yield_db_read_only_session),
 ) -> List[data.RegisteredContractResponse]:
     """
@@ -103,7 +98,7 @@ async def list_registered_contracts_route(
         registered_contracts_with_blockchain = (
             contracts_actions.lookup_registered_contracts(
                 db_session=db_session,
-                metatx_requester_id=request.state.user.id,
+                metatx_requester_id=user.id,
                 blockchain=blockchain,
                 address=address,
                 limit=limit,
@@ -126,8 +121,8 @@ async def list_registered_contracts_route(
     response_model=data.RegisteredContractResponse,
 )
 async def get_registered_contract_route(
-    request: Request,
     contract_id: UUID = Path(...),
+    user: BugoutUser = Depends(request_user_auth),
     db_session: Session = Depends(db.yield_db_read_only_session),
 ) -> List[data.RegisteredContractResponse]:
     """
@@ -136,7 +131,7 @@ async def get_registered_contract_route(
     try:
         contract_with_blockchain = contracts_actions.get_registered_contract(
             db_session=db_session,
-            metatx_requester_id=request.state.user.id,
+            metatx_requester_id=user.id,
             contract_id=contract_id,
         )
     except NoResultFound:
@@ -157,8 +152,8 @@ async def get_registered_contract_route(
     "/contracts", tags=["contracts"], response_model=data.RegisteredContractResponse
 )
 async def register_contract_route(
-    request: Request,
     contract: data.RegisterContractRequest = Body(...),
+    user: BugoutUser = Depends(request_user_auth),
     db_session: Session = Depends(db.yield_db_session),
 ) -> data.RegisteredContractResponse:
     """
@@ -167,7 +162,7 @@ async def register_contract_route(
     try:
         contract_with_blockchain = contracts_actions.register_contract(
             db_session=db_session,
-            metatx_requester_id=request.state.user.id,
+            metatx_requester_id=user.id,
             blockchain_name=contract.blockchain,
             address=contract.address,
             title=contract.title,
@@ -198,15 +193,15 @@ async def register_contract_route(
     response_model=data.RegisteredContractResponse,
 )
 async def update_contract_route(
-    request: Request,
     contract_id: UUID = Path(...),
     update_info: data.UpdateContractRequest = Body(...),
+    user: BugoutUser = Depends(request_user_auth),
     db_session: Session = Depends(db.yield_db_session),
 ) -> data.RegisteredContractResponse:
     try:
         contract_with_blockchain = contracts_actions.update_registered_contract(
             db_session=db_session,
-            metatx_requester_id=request.state.user.id,
+            metatx_requester_id=user.id,
             contract_id=contract_id,
             title=update_info.title,
             description=update_info.description,
@@ -233,8 +228,8 @@ async def update_contract_route(
     response_model=data.RegisteredContractResponse,
 )
 async def delete_contract_route(
-    request: Request,
     contract_id: UUID = Path(...),
+    user: BugoutUser = Depends(request_user_auth),
     db_session: Session = Depends(db.yield_db_session),
 ) -> data.RegisteredContractResponse:
     """
@@ -243,7 +238,7 @@ async def delete_contract_route(
     try:
         deleted_contract_with_blockchain = contracts_actions.delete_registered_contract(
             db_session=db_session,
-            metatx_requester_id=request.state.user.id,
+            metatx_requester_id=user.id,
             registered_contract_id=contract_id,
         )
     except Exception as err:
@@ -278,14 +273,20 @@ async def call_request_types_route(
     return call_request_types
 
 
-@app.get("/requests", tags=["requests"], response_model=List[data.CallRequestResponse])
+@app.get(
+    "/requests",
+    tags=["requests"],
+    response_model=List[data.CallRequestResponse],
+)
 async def list_requests_route(
     contract_id: Optional[UUID] = Query(None),
     contract_address: Optional[str] = Query(None),
     caller: str = Query(...),
     limit: int = Query(100),
     offset: Optional[int] = Query(None),
-    show_expired: Optional[bool] = Query(False),
+    show_expired: bool = Query(False),
+    live_after: Optional[int] = Query(None),
+    user: Optional[BugoutUser] = Depends(request_none_or_user_auth),
     db_session: Session = Depends(db.yield_db_read_only_session),
 ) -> List[data.CallRequestResponse]:
     """
@@ -302,6 +303,8 @@ async def list_requests_route(
             limit=limit,
             offset=offset,
             show_expired=show_expired,
+            live_after=live_after,
+            metatx_requester_id=user.id if user is not None else None,
         )
     except ValueError as e:
         logger.error(repr(e))
@@ -318,6 +321,7 @@ async def list_requests_route(
 )
 async def get_request(
     request_id: UUID = Path(...),
+    _: BugoutUser = Depends(request_user_auth),
     db_session: Session = Depends(db.yield_db_read_only_session),
 ) -> List[data.CallRequestResponse]:
     """
@@ -326,7 +330,7 @@ async def get_request(
     At least one of `contract_id` or `contract_address` must be provided as query parameters.
     """
     try:
-        request = contracts_actions.get_call_requests(
+        request = contracts_actions.get_call_request(
             db_session=db_session,
             request_id=request_id,
         )
@@ -344,8 +348,8 @@ async def get_request(
 
 @app.post("/requests", tags=["requests"], response_model=int)
 async def create_requests(
-    request: Request,
     data: data.CreateCallRequestsAPIRequest = Body(...),
+    user: BugoutUser = Depends(request_user_auth),
     db_session: Session = Depends(db.yield_db_session),
 ) -> int:
     """
@@ -354,13 +358,14 @@ async def create_requests(
     At least one of `contract_id` or `contract_address` must be provided in the request body.
     """
     try:
-        num_requests = contracts_actions.request_calls(
+        num_requests = contracts_actions.create_request_calls(
             db_session=db_session,
-            metatx_requester_id=request.state.user.id,
+            metatx_requester_id=user.id,
             registered_contract_id=data.contract_id,
             contract_address=data.contract_address,
             call_specs=data.specifications,
             ttl_days=data.ttl_days,
+            live_at=data.live_at,
         )
     except contracts_actions.InvalidAddressFormat as err:
         raise EngineHTTPException(
@@ -396,8 +401,8 @@ async def create_requests(
 
 @app.delete("/requests", tags=["requests"], response_model=int)
 async def delete_requests(
-    request: Request,
     request_ids: List[UUID] = Body(...),
+    user: BugoutUser = Depends(request_user_auth),
     db_session: Session = Depends(db.yield_db_session),
 ) -> int:
     """
@@ -406,7 +411,7 @@ async def delete_requests(
     try:
         deleted_requests = contracts_actions.delete_requests(
             db_session=db_session,
-            metatx_requester_id=request.state.user.id,
+            metatx_requester_id=user.id,
             request_ids=request_ids,
         )
     except Exception as err:
@@ -414,3 +419,32 @@ async def delete_requests(
         raise EngineHTTPException(status_code=500)
 
     return deleted_requests
+
+
+@app.post("/requests/{request_id}/complete", tags=["requests"])
+async def complete_call_request_route(
+    complete_request: data.CompleteCallRequestsAPIRequest = Body(...),
+    request_id: UUID = Path(...),
+    message=Depends(metatx_verify_header),
+    db_session: Session = Depends(db.yield_db_session),
+):
+    """
+    Set tx hash for specified call_request by verified account.
+    """
+    try:
+        request = contracts_actions.complete_call_request(
+            db_session=db_session,
+            tx_hash=complete_request.tx_hash,
+            call_request_id=request_id,
+            caller=message["caller"],
+        )
+    except contracts_actions.CallRequestNotFound:
+        raise EngineHTTPException(
+            status_code=404,
+            detail="There is no call request with that ID.",
+        )
+    except Exception as e:
+        logger.error(repr(e))
+        raise EngineHTTPException(status_code=500)
+
+    return contracts_actions.parse_call_request_response(request)

engineapi/engineapi/version.txt

@@ -1 +1 @@
-0.0.7
+0.0.8

engineapi/mypy.ini

@@ -0,0 +1,4 @@
+[mypy]
+
+[mypy-eth_keys.*]
+ignore_missing_imports = True

engineapi/requirements.txt

@@ -7,7 +7,7 @@ base58==2.1.1
 bitarray==2.7.6
 boto3==1.27.0
 botocore==1.30.0
-bugout==0.2.14
+bugout==0.2.15
 certifi==2023.5.7
 charset-normalizer==3.1.0
 click==8.1.3

engineapi/setup.py

@@ -13,7 +13,7 @@ setup(
     packages=find_packages(),
     install_requires=[
         "boto3",
-        "bugout>=0.2.14",
+        "bugout>=0.2.15",
         "eip712==0.1.0",
         "eth-typing>=2.3.0",
         "fastapi",

moonstreamapi/moonstreamapi/actions.py

@@ -15,6 +15,9 @@ from bugout.data import (
     BugoutResources,
     BugoutSearchResult,
     BugoutSearchResults,
+    BugoutResourceHolder,
+    HolderType,
+    ResourcePermissions,
 )
 from bugout.exceptions import BugoutResponseException
 from bugout.journal import SearchOrder
@@ -470,7 +473,7 @@ def upload_abi_to_s3(
 
 
 def get_all_entries_from_search(
-    journal_id: str, search_query: str, limit: int, token: str
+    journal_id: str, search_query: str, limit: int, token: str, content: bool = False
 ) -> List[BugoutSearchResult]:
     """
     Get all required entries from journal using search interface
@@ -483,7 +486,7 @@ def get_all_entries_from_search(
         token=token,
         journal_id=journal_id,
         query=search_query,
-        content=False,
+        content=content,
         timeout=10.0,
         limit=limit,
         offset=offset,
@@ -496,7 +499,7 @@ def get_all_entries_from_search(
                 token=token,
                 journal_id=journal_id,
                 query=search_query,
-                content=False,
+                content=content,
                 timeout=10.0,
                 limit=limit,
                 offset=offset,
@@ -526,47 +529,45 @@ def apply_moonworm_tasks(
             token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
         )
 
-        # create historical crawl task in journal
-
         # will use create_entries_pack for creating entries in journal
 
         existing_tags = [entry.tags for entry in entries]
 
-        existing_hashes = [
+        existing_selectors = [
-            tag.split(":")[-1]
+            tag.split(":")[-1] for tag in chain(*existing_tags) if "abi_selector" in tag
-            for tag in chain(*existing_tags)
-            if "abi_method_hash" in tag
         ]
 
-        abi_hashes_dict = {
+        abi_selectors_dict = {
-            hashlib.md5(json.dumps(method).encode("utf-8")).hexdigest(): method
+            Web3.keccak(
+                text=method["name"]
+                + "("
+                + ",".join(map(lambda x: x["type"], method["inputs"]))
+                + ")"
+            )[:4].hex(): method
             for method in abi
             if (method["type"] in ("event", "function"))
             and (method.get("stateMutability", "") != "view")
         }
 
-        for hash in abi_hashes_dict:
+        for abi_selector in abi_selectors_dict:
-            if hash not in existing_hashes:
+            if abi_selector not in existing_selectors:
-                abi_selector = Web3.keccak(
+                hash = hashlib.md5(
-                    text=abi_hashes_dict[hash]["name"]
+                    json.dumps(abi_selectors_dict[abi_selector]).encode("utf-8")
-                    + "("
+                ).hexdigest()
-                    + ",".join(
-                        map(lambda x: x["type"], abi_hashes_dict[hash]["inputs"])
-                    )
-                    + ")"
-                )[:4].hex()
 
                 moonworm_abi_tasks_entries_pack.append(
                     {
                         "title": address,
-                        "content": json.dumps(abi_hashes_dict[hash], indent=4),
+                        "content": json.dumps(
+                            abi_selectors_dict[abi_selector], indent=4
+                        ),
                         "tags": [
                             f"address:{address}",
-                            f"type:{abi_hashes_dict[hash]['type']}",
+                            f"type:{abi_selectors_dict[abi_selector]['type']}",
                             f"abi_method_hash:{hash}",
                             f"abi_selector:{abi_selector}",
                             f"subscription_type:{subscription_type}",
-                            f"abi_name:{abi_hashes_dict[hash]['name']}",
+                            f"abi_name:{abi_selectors_dict[abi_selector]['name']}",
                             f"status:active",
                             f"task_type:moonworm",
                             f"moonworm_task_pickedup:False",  # True if task picked up by moonworm-crawler(default each 120 sec)
@@ -711,11 +712,7 @@ def generate_journal_for_user(
     }
 
     try:
-        bc.create_resource(
+        create_resource_for_user(user_id=user_id, resource_data=resource_data)
-            token=token,
-            application_id=MOONSTREAM_APPLICATION_ID,
-            resource_data=resource_data,
-        )
     except BugoutResponseException as e:
         raise MoonstreamHTTPException(status_code=e.status_code, detail=e.detail)
     except Exception as e:
@@ -851,6 +848,8 @@ def get_list_of_support_interfaces(
     Returns list of interfaces supported by given address
     """
 
+    result = {}
+
     try:
         _, _, is_contract = check_if_smart_contract(
             blockchain_type=blockchain_type, address=address, user_token=user_token
@@ -866,8 +865,6 @@ def get_list_of_support_interfaces(
             abi=supportsInterface_abi,
         )
 
-        result = {}
-
         if blockchain_type in multicall_contracts:
             calls = []
 
@@ -952,3 +949,57 @@ def check_if_smart_contract(
         is_contract = True
 
     return blockchain_type, address, is_contract
+
+
+def create_resource_for_user(
+    user_id: uuid.UUID,
+    resource_data: Dict[str, Any],
+) -> BugoutResource:
+    """
+    Create resource for user
+    """
+    try:
+        resource = bc.create_resource(
+            token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
+            application_id=MOONSTREAM_APPLICATION_ID,
+            resource_data=resource_data,
+            timeout=BUGOUT_REQUEST_TIMEOUT_SECONDS,
+        )
+    except BugoutResponseException as e:
+        raise MoonstreamHTTPException(status_code=e.status_code, detail=e.detail)
+    except Exception as e:
+        logger.error(f"Error creating resource: {str(e)}")
+        raise MoonstreamHTTPException(status_code=500, internal_error=e)
+
+    try:
+        bc.add_resource_holder_permissions(
+            token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
+            resource_id=resource.id,
+            holder_permissions=BugoutResourceHolder(
+                holder_type=HolderType.user,
+                holder_id=user_id,
+                permissions=[
+                    ResourcePermissions.ADMIN,
+                    ResourcePermissions.READ,
+                    ResourcePermissions.UPDATE,
+                    ResourcePermissions.DELETE,
+                ],
+            ),
+            timeout=BUGOUT_REQUEST_TIMEOUT_SECONDS,
+        )
+    except BugoutResponseException as e:
+        logger.error(
+            f"Error adding resource holder permissions to resource resource {str(resource.id)}  {str(e)}"
+        )
+        raise MoonstreamHTTPException(status_code=e.status_code, detail=e.detail)
+    except Exception as e:
+        bc.delete_resource(
+            token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
+            resource_id=resource.id,
+        )
+        logger.error(
+            f"Error adding resource holder permissions to resource {str(resource.id)}  {str(e)}"
+        )
+        raise MoonstreamHTTPException(status_code=500, internal_error=e)
+
+    return resource

moonstreamapi/moonstreamapi/admin/cli.py

@@ -12,7 +12,12 @@ from sqlalchemy.orm import with_expression
 
 from moonstreamdb.db import SessionLocal
 
-from ..settings import BUGOUT_BROOD_URL, BUGOUT_SPIRE_URL, MOONSTREAM_APPLICATION_ID
+from ..settings import (
+    BUGOUT_BROOD_URL,
+    BUGOUT_SPIRE_URL,
+    MOONSTREAM_APPLICATION_ID,
+    MOONSTREAM_MOONWORM_TASKS_JOURNAL,
+)
 from ..web3_provider import yield_web3_provider
 
 from . import subscription_types, subscriptions, moonworm_tasks, queries
@@ -20,6 +25,7 @@ from .migrations import (
     checksum_address,
     update_dashboard_subscription_key,
     generate_entity_subscriptions,
+    add_selectors,
 )
 
 
@@ -87,6 +93,9 @@ steps:
 - id: 20230501
 name: fix_duplicates_keys_in_entity_subscription
 description: Fix entity duplicates keys for all subscriptions introduced in 20230213
+- id: 20230904
+name fill_missing_selectors_in_moonworm_tasks
+description: Get all moonworm jobs from moonworm journal and add selector tag if it not represent
     """
     logger.info(entity_migration_overview)
 
@@ -117,6 +126,30 @@ def migrations_run(args: argparse.Namespace) -> None:
     web3_session = yield_web3_provider()
     db_session = SessionLocal()
     try:
+        if args.id == 20230904:
+            step_order = [
+                "fill_missing_selectors_in_moonworm_tasks",
+                "deduplicate_moonworm_tasks",
+            ]
+            step_map: Dict[str, Dict[str, Any]] = {
+                "upgrade": {
+                    "fill_missing_selectors_in_moonworm_tasks": {
+                        "action": add_selectors.fill_missing_selectors_in_moonworm_tasks,
+                        "description": "Get all moonworm jobs from moonworm journal and add selector tag if it not represent",
+                    },
+                    "deduplicate_moonworm_tasks": {
+                        "action": add_selectors.deduplicate_moonworm_task_by_selector,
+                        "description": "Deduplicate moonworm tasks by selector",
+                    },
+                },
+                "downgrade": {},
+            }
+            if args.command not in ["upgrade", "downgrade"]:
+                logger.info("Wrong command. Please use upgrade or downgrade")
+            step = args.step
+
+            migration_run(step_map, args.command, step, step_order)
+
         if args.id == 20230501:
             # fix entity duplicates keys for all subscriptions introduced in 20230213
 

moonstreamapi/moonstreamapi/admin/migrations/add_selectors.py

@@ -0,0 +1,187 @@
+"""
+Add selectors to all moonworm tasks.
+"""
+import logging
+import json
+
+
+from bugout.exceptions import BugoutResponseException
+from web3 import Web3
+
+from ...settings import (
+    BUGOUT_REQUEST_TIMEOUT_SECONDS,
+    MOONSTREAM_ADMIN_ACCESS_TOKEN,
+    MOONSTREAM_MOONWORM_TASKS_JOURNAL,
+)
+from ...settings import bugout_client as bc
+from ...actions import get_all_entries_from_search
+
+logger = logging.getLogger(__name__)
+
+
+def fill_missing_selectors_in_moonworm_tasks() -> None:
+    """
+    Add selectors to all moonworm tasks.
+    """
+
+    batch_size = 100
+
+    moonworm_tasks = get_all_entries_from_search(
+        journal_id=MOONSTREAM_MOONWORM_TASKS_JOURNAL,
+        token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
+        search_query="#task_type:moonworm !#version:2.0",
+        limit=batch_size,
+        content=True,
+    )
+
+    logger.info(f"Found {len(moonworm_tasks)} moonworm tasks versions 1.0")
+
+    entries_tags = []
+
+    ## batch tasks
+
+    for task_batch in [
+        moonworm_tasks[i : i + batch_size]
+        for i in range(0, len(moonworm_tasks), batch_size)
+    ]:
+        count = 0
+        for task in task_batch:
+            tags = ["version:2.0"]
+
+            ## get abi
+            try:
+                abi = json.loads(task.content)
+            except Exception as e:
+                logger.warn(
+                    f"Unable to parse abi from task: {task.entry_url.split()[-1]}: {e}"
+                )
+                continue
+
+            if "name" not in abi:
+                logger.warn(
+                    f"Unable to find abi name in task: {task.entry_url.split()[-1]}"
+                )
+                continue
+
+            if not any([tag.startswith("abi_selector:") for tag in task.tags]):
+                ## generate selector
+
+                abi_selector = Web3.keccak(
+                    text=abi["name"]
+                    + "("
+                    + ",".join(map(lambda x: x["type"], abi["inputs"]))
+                    + ")"
+                )[:4].hex()
+
+                tags.append(f"abi_selector:{abi_selector}")
+
+                count += 1
+
+            entries_tags.append(
+                {
+                    "entry_id": task.entry_url.split("/")[-1],  ## 😭
+                    "tags": tags,
+                }
+            )
+
+        logger.info(f"Found {count} missing selectors in batch {len(task_batch)} tasks")
+
+        ## update entries
+
+        try:
+            bc.create_entries_tags(
+                journal_id=MOONSTREAM_MOONWORM_TASKS_JOURNAL,
+                token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
+                entries_tags=entries_tags,
+                timeout=15,
+            )
+        except BugoutResponseException as e:
+            logger.error(f"Unable to update entries tags: {e}")
+            continue
+
+
+def deduplicate_moonworm_task_by_selector():
+    """
+    Find moonworm tasks with same selector and remove old versions
+    """
+
+    moonworm_tasks = get_all_entries_from_search(
+        journal_id=MOONSTREAM_MOONWORM_TASKS_JOURNAL,
+        token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
+        search_query="#task_type:moonworm #version:2.0",
+        limit=100,
+        content=False,
+    )
+
+    logger.info(f"Found {len(moonworm_tasks)} moonworm tasks versions 2.0")
+
+    ## loop over tasks
+
+    selectors = {}
+
+    for task in moonworm_tasks:
+        tags = task.tags
+
+        ## get selector
+        selector = [tag for tag in tags if tag.startswith("abi_selector:")]
+
+        address = [tag for tag in tags if tag.startswith("address:")]
+
+        if len(selector) == 0:
+            logger.warn(
+                f"Unable to find selector in task: {task.entry_url.split()[-1]}"
+            )
+            continue
+
+        selector = selector[0].split(":")[1]
+
+        if len(address) == 0:
+            logger.warn(f"Unable to find address in task: {task.entry_url.split()[-1]}")
+            continue
+
+        address = address[0].split(":")[1]
+
+        if address not in selectors:
+            selectors[address] = {}
+
+        if selector not in selectors[address]:
+            selectors[address][selector] = {"entries": {}}
+
+        selectors[address][selector]["entries"][
+            task.entry_url.split("/")[-1]
+        ] = task.created_at
+
+    logger.info(f"Found {len(selectors)} addresses")
+
+    for address, selectors_dict in selectors.items():
+        for selector, tasks_dict in selectors_dict.items():
+            if len(tasks_dict["entries"]) == 1:
+                continue
+
+            ## find earliest task
+
+            earliest_task_id = min(
+                tasks_dict["entries"], key=lambda key: tasks_dict["entries"][key]
+            )
+
+            ## remove all tasks except latest
+
+            logger.info(
+                f"Found {len(tasks_dict['entries'])} tasks with selector {selector} erliest task {earliest_task_id} with created_at: {tasks_dict['entries'][earliest_task_id]}"
+            )
+
+            for task_id in tasks_dict["entries"]:
+                if task_id == earliest_task_id:
+                    continue
+
+                try:
+                    bc.delete_entry(
+                        journal_id=MOONSTREAM_MOONWORM_TASKS_JOURNAL,
+                        entry_id=task_id,
+                        token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
+                    )
+                except BugoutResponseException as e:
+                    logger.error(f"Unable to delete entry with id {task_id} : {e}")
+                    continue
+
+                logger.info(f"Deleted entry: {task_id}")

moonstreamapi/moonstreamapi/routes/queries.py

@@ -25,11 +25,11 @@ from ..actions import (
     get_query_by_name,
     name_normalization,
     query_parameter_hash,
+    create_resource_for_user,
 )
 from ..middleware import MoonstreamHTTPException
 from ..settings import (
     MOONSTREAM_ADMIN_ACCESS_TOKEN,
-    MOONSTREAM_APPLICATION_ID,
     MOONSTREAM_CRAWLERS_SERVER_PORT,
     MOONSTREAM_CRAWLERS_SERVER_URL,
     MOONSTREAM_INTERNAL_REQUEST_TIMEOUT_SECONDS,
@@ -130,24 +130,16 @@ async def create_query_handler(
     except Exception as e:
         raise MoonstreamHTTPException(status_code=500, internal_error=e)
 
-    try:
+    create_resource_for_user(
-        # create resource query_name_resolver
+        user_id=user.id,
-        bc.create_resource(
+        resource_data={
-            token=token,
+            "type": data.BUGOUT_RESOURCE_QUERY_RESOLVER,
-            application_id=MOONSTREAM_APPLICATION_ID,
+            "user_id": str(user.id),
-            resource_data={
+            "user": str(user.username),
-                "type": data.BUGOUT_RESOURCE_QUERY_RESOLVER,
+            "name": query_name,
-                "user_id": str(user.id),
+            "entry_id": str(entry.id),
-                "user": str(user.username),
+        },
-                "name": query_name,
+    )
-                "entry_id": str(entry.id),
-            },
-        )
-    except BugoutResponseException as e:
-        logger.error(f"Error creating name resolving resource: {str(e)}")
-        raise MoonstreamHTTPException(status_code=e.status_code, detail=e.detail)
-    except Exception as e:
-        raise MoonstreamHTTPException(status_code=500, internal_error=e)
 
     try:
         bc.update_tags(
@@ -355,7 +347,7 @@ async def update_query_handler(
             token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
             journal_id=MOONSTREAM_QUERIES_JOURNAL_ID,
             entry_id=query_id,
-            title=query_name,
+            title=f"Query:{query_name}",
             content=request_update.query,
             tags=["preapprove"],
         )
@@ -620,7 +612,9 @@ async def remove_query_handler(
         raise MoonstreamHTTPException(status_code=404, detail="Query does not exists")
 
     try:
-        bc.delete_resource(token=token, resource_id=query_ids[query_name][0])
+        bc.delete_resource(
+            token=MOONSTREAM_ADMIN_ACCESS_TOKEN, resource_id=query_ids[query_name][0]
+        )
     except BugoutResponseException as e:
         raise MoonstreamHTTPException(status_code=e.status_code, detail=e.detail)
     except Exception as e:

moonstreamapi/requirements.txt

@@ -9,7 +9,7 @@ base58==2.1.1
 bitarray==2.6.0
 boto3==1.26.5
 botocore==1.29.5
-bugout>=0.2.13
+bugout>=0.2.15
 certifi==2022.9.24
 charset-normalizer==2.1.1
 click==8.1.3

moonstreamapi/setup.py

@@ -13,7 +13,7 @@ setup(
     install_requires=[
         "appdirs",
         "boto3",
-        "bugout>=0.2.13",
+        "bugout>=0.2.15",
         "fastapi",
         "moonstreamdb>=0.3.5",
         "humbug",

nodebalancer/cmd/nodebalancer/balancer.go

@@ -198,7 +198,7 @@ func (bpool *BlockchainPool) HealthCheck() {
 	for _, b := range bpool.Blockchains {
 		var timeout time.Duration
 		getLatestBlockReq := `{"jsonrpc":"2.0","method":"eth_getBlockByNumber","params":["latest", false],"id":1}`
-		if b.Blockchain == "starknet" || b.Blockchain == "starknet-goerli" {
+		if b.Blockchain == "starknet" || b.Blockchain == "starknet-goerli" || b.Blockchain == "starknet-sepolia" {
 			getLatestBlockReq = `{"jsonrpc":"2.0","method":"starknet_getBlockWithTxHashes","params":["latest"],"id":"0"}`
 			timeout = NB_HEALTH_CHECK_CALL_TIMEOUT * 2
 		}
@@ -241,7 +241,7 @@ func (bpool *BlockchainPool) HealthCheck() {
 			}
 
 			var blockNumber uint64
-			if b.Blockchain == "starknet" || b.Blockchain == "starknet-goerli" {
+			if b.Blockchain == "starknet" || b.Blockchain == "starknet-goerli" || b.Blockchain == "starknet-sepolia" {
 				blockNumber = statusResponse.Result.BlockNumber
 			} else {
 				blockNumberHex := strings.Replace(statusResponse.Result.Number, "0x", "", -1)

nodebalancer/cmd/nodebalancer/configs.go

@@ -35,6 +35,7 @@ var (
 	NB_CONTROLLER_TOKEN             = os.Getenv("NB_CONTROLLER_TOKEN")
 	NB_CONTROLLER_ACCESS_ID         = os.Getenv("NB_CONTROLLER_ACCESS_ID")
 	MOONSTREAM_CORS_ALLOWED_ORIGINS = os.Getenv("MOONSTREAM_CORS_ALLOWED_ORIGINS")
+	CORS_WHITELIST_MAP              = make(map[string]bool)
 
 	NB_CONNECTION_RETRIES          = 2
 	NB_CONNECTION_RETRIES_INTERVAL = time.Millisecond * 10
@@ -86,6 +87,9 @@ func CheckEnvVarSet() {
 		NB_CONTROLLER_ACCESS_ID = uuid.New().String()
 		log.Printf("Access ID for internal usage in NB_CONTROLLER_ACCESS_ID environment variable is not valid uuid, generated random one: %v", NB_CONTROLLER_ACCESS_ID)
 	}
+	for _, o := range strings.Split(MOONSTREAM_CORS_ALLOWED_ORIGINS, ",") {
+		CORS_WHITELIST_MAP[o] = true
+	}
 }
 
 // Nodes configuration

nodebalancer/cmd/nodebalancer/middleware.go

@@ -368,19 +368,29 @@ func panicMiddleware(next http.Handler) http.Handler {
 // CORS middleware
 func corsMiddleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.Method == http.MethodOptions {
+		var allowedOrigin string
-			for _, allowedOrigin := range strings.Split(MOONSTREAM_CORS_ALLOWED_ORIGINS, ",") {
+		if CORS_WHITELIST_MAP["*"] {
-				if r.Header.Get("Origin") == allowedOrigin {
+			allowedOrigin = "*"
-					w.Header().Set("Access-Control-Allow-Origin", allowedOrigin)
+		} else {
-					w.Header().Set("Access-Control-Allow-Methods", "GET,POST")
+			origin := r.Header.Get("Origin")
-					// Credentials are cookies, authorization headers, or TLS client certificates
+			if _, ok := CORS_WHITELIST_MAP[origin]; ok {
-					w.Header().Set("Access-Control-Allow-Credentials", "true")
+				allowedOrigin = origin
-					w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type")
-				}
 			}
-			w.WriteHeader(http.StatusNoContent)
+		}
+
+		if allowedOrigin != "" {
+			w.Header().Set("Access-Control-Allow-Origin", allowedOrigin)
+			w.Header().Set("Access-Control-Allow-Methods", "GET,POST,OPTIONS")
+			// Credentials are cookies, authorization headers, or TLS client certificates
+			w.Header().Set("Access-Control-Allow-Credentials", "true")
+			w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type")
+		}
+
+		if r.Method == "OPTIONS" {
+			w.WriteHeader(http.StatusOK)
 			return
 		}
+
 		next.ServeHTTP(w, r)
 	})
 }

nodebalancer/cmd/nodebalancer/server.go

@@ -31,7 +31,7 @@ var (
 func initHealthCheck(debug bool) {
 	healthCheckInterval, convErr := strconv.Atoi(NB_HEALTH_CHECK_INTERVAL)
 	if convErr != nil {
-		healthCheckInterval = 5
+		healthCheckInterval = 30
 	}
 	t := time.NewTicker(time.Second * time.Duration(healthCheckInterval))
 	for {