diff --git a/.gitignore b/.gitignore index 4e8949f..734b6e9 100644 --- a/.gitignore +++ b/.gitignore @@ -147,6 +147,7 @@ external_packages/** */config.py config.local.json +config.dev.json /package.json /requirements.txt /serverless.yml \ No newline at end of file diff --git a/airdrop/application/handlers/airdrop_handlers.py b/airdrop/application/handlers/airdrop_handlers.py index 82d6feb..aceb17f 100644 --- a/airdrop/application/handlers/airdrop_handlers.py +++ b/airdrop/application/handlers/airdrop_handlers.py @@ -3,7 +3,7 @@ sys.path.append('/opt') from common.exception_handler import exception_handler -from airdrop.config import SLACK_HOOK, NETWORK_ID +from airdrop.config import MATTERMOST_CONFIG, NETWORK_ID from common.logger import get_logger from common.utils import generate_lambda_response, request from airdrop.application.services.airdrop_services import AirdropServices @@ -14,7 +14,7 @@ logger = get_logger(__name__) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def get_airdrop_schedules(event, context): logger.info(f"Got Airdrops Event {event}") parameters = event['pathParameters'] @@ -28,7 +28,7 @@ def get_airdrop_schedules(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def user_registration(event, context): logger.info(f"Got Airdrops Event {event}") status, response = UserRegistrationServices().register(request(event)) @@ -40,7 +40,7 @@ def user_registration(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def user_registration_update(event, context): logger.info(f"Got Airdrops Event {event}") status, response = UserRegistrationServices().update_registration(request(event)) @@ -52,7 +52,7 @@ def user_registration_update(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def airdrop_window_stake_details(event, context): logger.info(f"Got Airdrops Window Stake details {event}") status, response = AirdropServices().get_airdrop_window_stake_details(request(event)) @@ -64,7 +64,7 @@ def airdrop_window_stake_details(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def address_eligibility(event, context): logger.info(f"Got Airdrops Event {event}") status, response = UserRegistrationServices().eligibility_v2(request(event)) @@ -76,7 +76,7 @@ def address_eligibility(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def user_eligibility(event, context): logger.info(f"Got Airdrops Event {event}") status, response = UserRegistrationServices().eligibility(request(event)) @@ -88,7 +88,7 @@ def user_eligibility(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def occam_airdrop_window_claim(event, context): logger.info(f"Got occam_airdrop_window_claim event {event}") status, response = AirdropServices().occam_airdrop_window_claim(request(event)) @@ -100,7 +100,7 @@ def occam_airdrop_window_claim(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def airdrop_window_claim(event, context): logger.info(f"Got airdrop_window_claim Events {event}") status, response = AirdropServices().airdrop_window_claim(request(event)) @@ -112,7 +112,7 @@ def airdrop_window_claim(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def airdrop_window_claim_status(event, context): logger.info(f"Got Airdrops Window Claims Statys Events {event}") status, response = AirdropServices( @@ -125,7 +125,7 @@ def airdrop_window_claim_status(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def airdrop_window_claim_history(event, context): logger.info(f"Got Airdrops Window Claims Statys Events {event}") status, response = AirdropServices( @@ -138,7 +138,7 @@ def airdrop_window_claim_history(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def airdrop_event_consumer(event, context): logger.info(f"Got Airdrops event listener {event}") status, response = AirdropServices( @@ -151,13 +151,13 @@ def airdrop_event_consumer(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def airdrop_txn_watcher(event, context): logger.info(f"Got Airdrops txn status watcher {event}") AirdropServices().airdrop_txn_watcher() -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger) def user_notifications(event, context): logger.info(f"Got Airdrops user notifications {event}") status, response = UserNotificationService( diff --git a/airdrop/application/handlers/consumer_handler.py b/airdrop/application/handlers/consumer_handler.py index e12168b..e61d211 100644 --- a/airdrop/application/handlers/consumer_handler.py +++ b/airdrop/application/handlers/consumer_handler.py @@ -1,19 +1,21 @@ import sys sys.path.append('/opt') -from airdrop.config import SLACK_HOOK, NETWORK_ID +from airdrop.config import MATTERMOST_CONFIG, NETWORK_ID from airdrop.application.services.event_consumer_service import DepositEventConsumerService from common.logger import get_logger from common.utils import generate_lambda_response from common.exception_handler import exception_handler -from common.exceptions import ValidationFailedException +from common.exceptions import BlockConfirmationException, ValidationFailedException logger = get_logger(__name__) -EXCEPTIONS = (ValidationFailedException,) +NOT_RAISED_CUSTOM_EXCEPTIONS = (ValidationFailedException,) +RAISED_CUSTOM_EXCEPTIONS = (BlockConfirmationException,) -@exception_handler(SLACK_HOOK=SLACK_HOOK, NETWORK_ID=NETWORK_ID, logger=logger, EXCEPTIONS=EXCEPTIONS, - RAISE_EXCEPTION=True) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, NETWORK_ID=NETWORK_ID, logger=logger, + NOT_RAISED_EXCEPTIONS=NOT_RAISED_CUSTOM_EXCEPTIONS, + RAISED_EXCEPTIONS=RAISED_CUSTOM_EXCEPTIONS, RAISE_EXCEPTION=True) def deposit_event_consumer(event, context): logger.info(f"Got deposit event {event}") response = DepositEventConsumerService(event).validate_deposit_event() diff --git a/airdrop/application/services/airdrop_services.py b/airdrop/application/services/airdrop_services.py index bf32220..755303b 100644 --- a/airdrop/application/services/airdrop_services.py +++ b/airdrop/application/services/airdrop_services.py @@ -7,28 +7,32 @@ import web3 from eth_account.messages import encode_defunct from jsonschema import validate, ValidationError +from sqlalchemy.exc import NoResultFound from web3 import Web3, types -from airdrop.config import NETWORK, DEFAULT_REGION from airdrop.config import ( - SIGNER_PRIVATE_KEY, - SIGNER_PRIVATE_KEY_STORAGE_REGION, - NUNET_SIGNER_PRIVATE_KEY_STORAGE_REGION, + DEFAULT_REGION, + MATTERMOST_CONFIG, + NETWORK, NUNET_SIGNER_PRIVATE_KEY, - SLACK_HOOK + NUNET_SIGNER_PRIVATE_KEY_STORAGE_REGION, + SIGNER_PRIVATE_KEY, + SIGNER_PRIVATE_KEY_STORAGE_REGION ) from airdrop.constants import ( - STAKING_CONTRACT_PATH, CLAIM_SCHEMA, - AirdropEvents, + PROCESSOR_PATH, + STAKING_CONTRACT_PATH, AirdropClaimStatus, - PROCESSOR_PATH + AirdropEvents, + Blockchain ) from airdrop.domain.factory.airdrop_factory import AirdropFactory from airdrop.infrastructure.repositories.airdrop_repository import AirdropRepository from airdrop.infrastructure.repositories.airdrop_window_repository import AirdropWindowRepository from airdrop.processor.default_airdrop import DefaultAirdrop, BaseAirdrop -from airdrop.utils import Utils as ut +from airdrop.utils import Utils +from common.alerts import MattermostProcessor from common.boto_utils import BotoUtils from common.logger import get_logger from common.utils import ( @@ -36,10 +40,10 @@ generate_claim_signature_with_total_eligibile_amount, get_contract_instance, get_transaction_receipt_from_blockchain, - get_checksum_address, - Utils + get_checksum_address ) +alert_processor = MattermostProcessor(config=MATTERMOST_CONFIG) logger = get_logger(__name__) @@ -177,10 +181,10 @@ def get_stake_info(self, contract_address, user_wallet_address, airdrop_rewards, except BaseException as e: logger.error(e) - Utils().report_slack( - type=0, slack_message=f"Issue with Stake window Opened exeption {e} user_address {user_wallet_address}," - f" stake_contract_address: {contract_address}", slack_config=SLACK_HOOK - ) + alert_processor.send(type=1, + message=f"Issue with Stake window Opened exeption {e} " + f"user_address {user_wallet_address}, " + f"stake_contract_address: {contract_address}") return False, 0, airdrop_rewards def get_stake_and_claimable_amounts(self, airdrop_rewards, is_stake_window_is_open, max_stake_amount, @@ -321,7 +325,7 @@ def airdrop_window_claim_status(self, inputs): amount = inputs["amount"] blockchain_method = inputs["blockchain_method"] - if ut.recognize_blockchain_network(user_address) == "Ethereum": + if Utils.recognize_blockchain_network(user_address) == Blockchain.ETHEREUM.value: user_address = Web3.to_checksum_address(user_address) AirdropRepository().airdrop_window_claim_txn( @@ -479,12 +483,12 @@ def get_signature_for_airdrop_window_id_with_totaleligibilty_amount(self, amount raise e def get_airdrops_schedule(self, airdrop_id): - status = HTTPStatus.BAD_REQUEST try: response = AirdropRepository().get_airdrops_schedule(airdrop_id) status = HTTPStatus.OK - except ValidationError as e: - response = e.message + except NoResultFound as e: + response = f"Airdrop with id = {airdrop_id} does not exist" + status = HTTPStatus.BAD_REQUEST except BaseException as e: response = str(e) status = HTTPStatus.INTERNAL_SERVER_ERROR diff --git a/airdrop/application/services/common_logic_service.py b/airdrop/application/services/common_logic_service.py new file mode 100644 index 0000000..8497dbf --- /dev/null +++ b/airdrop/application/services/common_logic_service.py @@ -0,0 +1,33 @@ +from typing import Optional, Tuple, Union + +from airdrop.constants import CARDANO_ADDRESS_PREFIXES, Blockchain, CardanoEra +from airdrop.infrastructure.models import UserRegistration +from airdrop.infrastructure.repositories.user_registration_repo import UserRegistrationRepository +from airdrop.utils import Utils + + +class CommonLogicService: + + @staticmethod + def get_user_registration_details( + address: str, + airdrop_window_id: int, + registration_id: Optional[str] = None + ) -> Tuple[bool, Optional[Union[UserRegistration, list[UserRegistration]]]]: + registration_repo = UserRegistrationRepository() + network = Utils.recognize_blockchain_network(address) + if (network == Blockchain.ETHEREUM.value or + address.startswith(tuple(CARDANO_ADDRESS_PREFIXES[CardanoEra.BYRON]))): + return registration_repo.get_user_registration_details( + address=address, + airdrop_window_id=airdrop_window_id, + registration_id=registration_id + ) + elif network == Blockchain.CARDANO.value: + payment_part, staking_part = Utils.get_payment_staking_parts(address) + return registration_repo.get_user_registration_details( + payment_part=payment_part, + staking_part=staking_part, + airdrop_window_id=airdrop_window_id, + registration_id=registration_id + ) diff --git a/airdrop/application/services/event_consumer_service.py b/airdrop/application/services/event_consumer_service.py index e469048..af4a71e 100644 --- a/airdrop/application/services/event_consumer_service.py +++ b/airdrop/application/services/event_consumer_service.py @@ -10,7 +10,7 @@ from airdrop.infrastructure.repositories.airdrop_repository import AirdropRepository from airdrop.infrastructure.repositories.airdrop_window_repository import AirdropWindowRepository from airdrop.infrastructure.repositories.user_registration_repo import UserRegistrationRepository -from common.exceptions import ValidationFailedException +from common.exceptions import BlockConfirmationException, ValidationFailedException from common.logger import get_logger user_registration_repo = UserRegistrationRepository() @@ -51,7 +51,7 @@ def validate_block_confirmation(self, transaction_block_no): current_block_no = EventConsumerService.get_current_block_no() if current_block_no > (transaction_block_no + MIN_BLOCK_CONFIRMATION_REQUIRED): return None - raise Exception(f"Block confirmation is not enough for event {self.event}") + raise BlockConfirmationException(f"Block confirmation is not enough for event {self.event}") @staticmethod def get_stake_key_address(address): diff --git a/airdrop/application/services/user_claim_service.py b/airdrop/application/services/user_claim_service.py index 3b66cb5..2d2c08e 100644 --- a/airdrop/application/services/user_claim_service.py +++ b/airdrop/application/services/user_claim_service.py @@ -5,16 +5,16 @@ import requests from airdrop.application.services.airdrop_services import AirdropServices -from airdrop.config import TokenTransferCardanoService, SLACK_HOOK, MIN_BLOCK_CONFIRMATION_REQUIRED +from airdrop.config import MATTERMOST_CONFIG, TokenTransferCardanoService, MIN_BLOCK_CONFIRMATION_REQUIRED from airdrop.constants import AirdropClaimStatus from airdrop.infrastructure.repositories.airdrop_repository import AirdropRepository from airdrop.infrastructure.repositories.claim_history_repo import ClaimHistoryRepository from airdrop.application.services.event_consumer_service import EventConsumerService +from common.alerts import MattermostProcessor from common.logger import get_logger -from common.utils import Utils +alert_processor = MattermostProcessor(config=MATTERMOST_CONFIG) logger = get_logger(__name__) -utils = Utils() class UserClaimService: @@ -54,7 +54,7 @@ def invoke_token_transfer_cardano_service(payload): f'{response_body.get("error", {}).get("message", "")}\nDetails ' \ f'{response_body.get("error", {}).get("details", "")}.' logger.exception(error_message) - utils.report_slack(type=1, slack_message=error_message, slack_config=SLACK_HOOK) + alert_processor.send(type=1, message=error_message) return {} return response_body @@ -86,7 +86,7 @@ def initiate_claim_for_users(self): transaction_status = AirdropClaimStatus.CLAIM_SUBMITTED.value else: transaction_status = AirdropClaimStatus.CLAIM_FAILED.value - utils.report_slack(type=0, slack_message="Token Transfer Cardano Service Failed!", slack_config=SLACK_HOOK) + alert_processor.send(type=1, message="Token Transfer Cardano Service Failed!") transaction_id = None # Update claim status as CLAIM_FAILED/CLAIM_SUBMITTED diff --git a/airdrop/application/services/user_registration_services.py b/airdrop/application/services/user_registration_services.py index 8feceaa..ca70727 100644 --- a/airdrop/application/services/user_registration_services.py +++ b/airdrop/application/services/user_registration_services.py @@ -1,15 +1,15 @@ -from datetime import datetime +import datetime from http import HTTPStatus from typing import List -from jsonschema import validate, ValidationError from blockfrost import BlockFrostApi from blockfrost.utils import ApiError as BlockFrostApiError -from web3 import Web3 +from jsonschema import validate, ValidationError from pycardano import Address from airdrop.application.services.airdrop_services import AirdropServices +from airdrop.application.services.common_logic_service import CommonLogicService from airdrop.config import BlockFrostAPIBaseURL, BlockFrostAccountDetails from airdrop.constants import ( CARDANO_ADDRESS_PREFIXES, @@ -17,11 +17,12 @@ ADDRESS_ELIGIBILITY_SCHEMA, USER_REGISTRATION_SCHEMA, AirdropClaimStatus, + AirdropEvents, CardanoEra, UserClaimStatus ) from airdrop.application.types.windows import WindowRegistrationData, RegistrationDetails -from airdrop.infrastructure.models import PendingTransaction +from airdrop.infrastructure.models import AirdropWindow, PendingTransaction from airdrop.infrastructure.repositories.airdrop_repository import AirdropRepository from airdrop.infrastructure.repositories.airdrop_window_repository import AirdropWindowRepository from airdrop.infrastructure.repositories.pending_transaction_repo import PendingTransactionRepository @@ -40,6 +41,7 @@ class UserRegistrationServices: def __generate_user_claim_status( is_registered: bool, airdrop_claim_status: AirdropClaimStatus | None, + airdrop_state_status: AirdropEvents ) -> UserClaimStatus: logger.debug( f"Generate user claim status" @@ -58,6 +60,8 @@ def __generate_user_claim_status( return UserClaimStatus.PENDING elif airdrop_claim_status == AirdropClaimStatus.NOT_STARTED: return UserClaimStatus.NOT_STARTED + elif airdrop_claim_status is None and airdrop_state_status == AirdropEvents.AIRDROP_WINDOW_OPEN: + return UserClaimStatus.REGISTERED elif airdrop_claim_status in ( AirdropClaimStatus.FAILED, None @@ -68,23 +72,24 @@ def __generate_user_claim_status( raise Exception(f"Unexpected aidrop_claim_status: {airdrop_claim_status}") @staticmethod - def __get_registration_data(address: str, airdrop_window_id: int) -> WindowRegistrationData: - is_registered, user_registration = UserRegistrationRepository().get_user_registration_details( - address, airdrop_window_id + def __get_registration_data(address: str, airdrop_window: AirdropWindow) -> WindowRegistrationData: + is_registered, user_registration = CommonLogicService.get_user_registration_details( + address=address, + airdrop_window_id=airdrop_window.id ) if isinstance(user_registration, list): - logger.error(f"Find multiple registrations for {address=}, {airdrop_window_id=}") + logger.error(f"Find multiple registrations for {address=}, {airdrop_window.id=}") raise BadRequestException("Something wrong with user registration") last_claim = ClaimHistoryRepository().get_last_claim_history( - airdrop_window_id=airdrop_window_id, + airdrop_window_id=airdrop_window.id, address=address, blockchain_method="token_transfer" ) last_ada_transfer = ClaimHistoryRepository().get_last_claim_history( - airdrop_window_id=airdrop_window_id, + airdrop_window_id=airdrop_window.id, address=address, blockchain_method="ada_transfer" ) @@ -95,7 +100,20 @@ def __get_registration_data(address: str, airdrop_window_id: int) -> WindowRegis elif last_ada_transfer is not None: airdrop_claim_status = AirdropClaimStatus(last_ada_transfer.transaction_status) - user_claim_status = UserRegistrationServices.__generate_user_claim_status(is_registered, airdrop_claim_status) + now = datetime_in_utcnow() + claim_start_period = airdrop_window.claim_start_period.replace(tzinfo=datetime.timezone.utc) + + airdrop_state_status = None + if now < claim_start_period: + airdrop_state_status = AirdropEvents.AIRDROP_WINDOW_OPEN + elif now >= claim_start_period: + airdrop_state_status = AirdropEvents.AIRDROP_CLAIM + + user_claim_status = UserRegistrationServices.__generate_user_claim_status( + is_registered=is_registered, + airdrop_claim_status=airdrop_claim_status, + airdrop_state_status=airdrop_state_status + ) registration_details = RegistrationDetails( registration_id = str(user_registration.receipt_generated), @@ -105,7 +123,7 @@ def __get_registration_data(address: str, airdrop_window_id: int) -> WindowRegis ) if is_registered and user_registration is not None else None window_registration_data = WindowRegistrationData( - window_id=airdrop_window_id, + window_id=airdrop_window.id, airdrop_window_claim_status=airdrop_claim_status, claim_status=user_claim_status, registration_details=registration_details @@ -126,9 +144,6 @@ def eligibility_v2(inputs: dict) -> tuple: wallet_name = inputs.get("wallet_name") key = inputs.get("key") - if Utils.recognize_blockchain_network(address) == "Ethereum": - address = Web3.to_checksum_address(address) - airdrop = AirdropRepository().get_airdrop_details(airdrop_id) if not airdrop: logger.error("Airdrop id is not valid") @@ -142,6 +157,8 @@ def eligibility_v2(inputs: dict) -> tuple: airdrop_class = AirdropServices.load_airdrop_class(airdrop) airdrop_object = airdrop_class(airdrop_id) + address = airdrop_object.to_checksum_address_if_ethereum(address) + is_user_eligible = airdrop_object.check_user_eligibility(address) with_signature = False @@ -161,7 +178,7 @@ def eligibility_v2(inputs: dict) -> tuple: windows_registration_data.append( UserRegistrationServices.__get_registration_data( address=address, - airdrop_window_id=window.id, + airdrop_window=window, ) ) @@ -205,8 +222,10 @@ def eligibility(inputs: dict) -> tuple: rewards_awarded = AirdropRepository().fetch_total_rewards_amount(airdrop_id, address) - is_registered, user_registration = UserRegistrationRepository(). \ - get_user_registration_details(address, airdrop_window_id) + is_registered, user_registration = CommonLogicService.get_user_registration_details( + address, + airdrop_window_id + ) is_airdrop_window_claimed = False is_claimable = False @@ -318,8 +337,7 @@ def check_trezor_registrations() -> None: registration_repo = UserRegistrationRepository() logger.info(f"Found tx {registration.tx_hash}: block={tx_data.block_height} index={tx_data.index}") - is_registered, _ = registration_repo.get_user_registration_details(registration.address, - registration.airdrop_window_id) + is_registered, _ = CommonLogicService.get_user_registration_details(registration.address) if is_registered: logger.error("Address is already registered for this airdrop window") raise Exception("Address is already registered for this airdrop window") @@ -331,7 +349,7 @@ def check_trezor_registrations() -> None: is_address_match = True break # Metadata check - is_metadata_match, metadata = Utils().compare_data_from_db_and_metadata( + is_metadata_match, metadata = Utils.compare_data_from_db_and_metadata( registration.signature_details, tx_metadata ) @@ -355,7 +373,7 @@ def check_trezor_registrations() -> None: for registration in to_save: payment_part: str | None = None staking_part: str | None = None - if any(registration.address.startswith(prefix) for prefix in CARDANO_ADDRESS_PREFIXES[CardanoEra.SHELLEY]): + if registration.address.startswith(tuple(CARDANO_ADDRESS_PREFIXES[CardanoEra.SHELLEY])): formatted_address = Address.from_primitive(registration.address) payment_part = str(formatted_address.payment_part) if formatted_address.payment_part else None staking_part = str(formatted_address.staking_part) if formatted_address.staking_part else None diff --git a/airdrop/config.example.py b/airdrop/config.example.py index 537a785..513a041 100644 --- a/airdrop/config.example.py +++ b/airdrop/config.example.py @@ -37,13 +37,9 @@ NETWORK_ID = 11155111 DEFAULT_REGION = "us-east-1" -SLACK_HOOK = { - "hostname": "https://hooks.slack.com", - "port": 443, - "path": "", - "method": "POST", - "headers": {"Content-Type": "application/json"}, - "channel_name": "airdrop-ropsten-alerts" + +MATTERMOST_CONFIG = { + "url": "https://chat.mattermost.io/hooks/test" } SIGNER_PRIVATE_KEY = 'AIRDROP_SIGNER_PRIVATE_KEY' diff --git a/airdrop/constants.py b/airdrop/constants.py index e2c0d93..54c765b 100644 --- a/airdrop/constants.py +++ b/airdrop/constants.py @@ -131,6 +131,7 @@ class UserClaimStatus(Enum): NOT_REGISTERED = "NOT_REGISTERED" NOT_STARTED = "NOT_STARTED" PENDING = "PENDING" + REGISTERED = "REGISTERED" class AirdropClaimStatus(Enum): @@ -161,6 +162,11 @@ class TransactionType(Enum): REGISTRATION = "Registration" +class Blockchain(Enum): + CARDANO = "Cardano" + ETHEREUM = "Ethereum" + + class CardanoEra(Enum): BYRON = "Byron" SHELLEY = "Shelley" diff --git a/airdrop/infrastructure/repositories/airdrop_repository.py b/airdrop/infrastructure/repositories/airdrop_repository.py index 9c307b9..674f35e 100644 --- a/airdrop/infrastructure/repositories/airdrop_repository.py +++ b/airdrop/infrastructure/repositories/airdrop_repository.py @@ -1,5 +1,5 @@ from sqlalchemy import text -from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.exc import NoResultFound, SQLAlchemyError from airdrop.constants import AirdropClaimStatus from airdrop.domain.factory.airdrop_factory import AirdropFactory @@ -376,16 +376,13 @@ def get_airdrops_schedule(self, airdrop_id): isouter = True ) .filter(Airdrop.id == airdrop_id) - .first() + .one() ) self.session.commit() - except SQLAlchemyError as e: + except (SQLAlchemyError, NoResultFound) as e: self.session.rollback() raise e - if airdrop_row_data is not None: - return AirdropFactory.convert_airdrop_schedule_model_to_entity_model(airdrop_row_data) - else: - raise Exception('Non eligible user') + return AirdropFactory.convert_airdrop_schedule_model_to_entity_model(airdrop_row_data) def get_airdrop_details(self, airdrop_id): return self.session.query(Airdrop).filter(Airdrop.id == airdrop_id).first() diff --git a/airdrop/infrastructure/repositories/airdrop_window_repository.py b/airdrop/infrastructure/repositories/airdrop_window_repository.py index f6f3bcd..2ee1485 100644 --- a/airdrop/infrastructure/repositories/airdrop_window_repository.py +++ b/airdrop/infrastructure/repositories/airdrop_window_repository.py @@ -37,6 +37,18 @@ def is_open_airdrop_window(self, airdrop_id, airdrop_window_id, date_time): .first() ) + def is_claimable_airdrop(self, airdrop_id, date_time): + """ + Rejuve Airdrop specific repository method + Indicates whether there is any airdrop window with an open claim period + """ + claimable_windows = self.session.query(AirdropWindow) \ + .filter(AirdropWindow.airdrop_id == airdrop_id, + AirdropWindow.claim_start_period <= date_time, + AirdropWindow.claim_end_period >= date_time) \ + .all() + return len(claimable_windows) > 0 + def get_airdrop_windows(self, airdrop_id: int) -> List[AirdropWindow]: return self.session.query(AirdropWindow) \ .filter(AirdropWindow.airdrop_id == airdrop_id) \ diff --git a/airdrop/infrastructure/repositories/base_repository.py b/airdrop/infrastructure/repositories/base_repository.py index 72c8ce2..0af7861 100644 --- a/airdrop/infrastructure/repositories/base_repository.py +++ b/airdrop/infrastructure/repositories/base_repository.py @@ -1,16 +1,18 @@ from sqlalchemy import create_engine +from sqlalchemy.engine import URL from sqlalchemy.orm import sessionmaker from airdrop.config import NETWORK -driver=NETWORK['db']['DB_DRIVER'] -host=NETWORK['db']['DB_HOST'] -user=NETWORK['db']["DB_USER"] -db_name=NETWORK['db']["DB_NAME"] -password=NETWORK['db']["DB_PASSWORD"] -port=NETWORK['db']["DB_PORT"] -connection_string = f"{driver}://{user}:{password}@{host}:{port}/{db_name}" -engine = create_engine(connection_string, pool_pre_ping=True, echo=False, isolation_level="READ COMMITTED") +url = URL.create( + drivername=NETWORK['db']['DB_DRIVER'], + username=NETWORK['db']["DB_USER"], + password=NETWORK['db']["DB_PASSWORD"], + host=NETWORK['db']['DB_HOST'], + port=NETWORK['db']["DB_PORT"], + database=NETWORK['db']["DB_NAME"] +) +engine = create_engine(url, pool_pre_ping=True, echo=False, isolation_level="READ COMMITTED") Session = sessionmaker(bind=engine) default_session = Session() diff --git a/airdrop/infrastructure/repositories/user_registration_repo.py b/airdrop/infrastructure/repositories/user_registration_repo.py index 45796cf..6eb0dec 100644 --- a/airdrop/infrastructure/repositories/user_registration_repo.py +++ b/airdrop/infrastructure/repositories/user_registration_repo.py @@ -4,8 +4,7 @@ from sqlalchemy.exc import SQLAlchemyError from airdrop.constants import AirdropClaimStatus -from airdrop.domain.factory.airdrop_factory import AirdropFactory -from airdrop.infrastructure.models import AirdropWindow, UserRegistration, UserReward, UserNotifications +from airdrop.infrastructure.models import UserRegistration, UserReward, UserNotifications from airdrop.infrastructure.repositories.base_repository import BaseRepository from airdrop.utils import datetime_in_utcnow @@ -27,27 +26,6 @@ def subscribe_to_notifications(self, email, airdrop_id): self.session.rollback() raise e - def airdrop_window_user_details(self, airdrop_window_id, address): - user_data = ( - self.session.query(UserRegistration) - .join( - AirdropWindow, - AirdropWindow.id == UserRegistration.airdrop_window_id, - ) - .filter(UserRegistration.airdrop_window_id == airdrop_window_id) - .filter(UserRegistration.address == address) - .filter(UserRegistration.registered_at != None) - .first() - ) - - user_details = None - - if user_data is not None: - user_details = AirdropFactory.convert_airdrop_window_user_model_to_entity_model( - user_data) - - return user_details - def get_reject_reason(self, airdrop_window_id, address): registration = ( self.session.query(UserRegistration.reject_reason) @@ -147,21 +125,41 @@ def is_user_eligible_for_given_window(self, address, airdrop_id, airdrop_window_ def get_user_registration_details( self, address: Optional[str] = None, + payment_part: Optional[str] = None, + staking_part: Optional[str] = None, airdrop_window_id: Optional[int] = None, registration_id: Optional[str] = None ) -> Tuple[bool, Optional[Union[UserRegistration, list[UserRegistration]]]]: - query = self.session.query(UserRegistration).filter(UserRegistration.registered_at != None) - if address: - query = query.filter(UserRegistration.address == address) - if airdrop_window_id: - query = query.filter(UserRegistration.airdrop_window_id == airdrop_window_id) - if registration_id: - query = query.filter(UserRegistration.receipt_generated == registration_id) - user_registrations = query.all() - registration_count = len(user_registrations) - if registration_count: - return True, user_registrations[0] if registration_count == 1 else user_registrations - return False, None + try: + or_clause = list() + if payment_part: + or_clause.append(UserRegistration.payment_part == payment_part) + if staking_part: + or_clause.append(UserRegistration.staking_part == staking_part) + if address: + or_clause.append(UserRegistration.address == address) + + query = ( + self.session.query(UserRegistration) + .filter( + UserRegistration.registered_at != None, + or_(*or_clause) + ) + ) + + if airdrop_window_id: + query = query.filter(UserRegistration.airdrop_window_id == airdrop_window_id) + if registration_id: + query = query.filter(UserRegistration.receipt_generated == registration_id) + + user_registrations = query.all() + registration_count = len(user_registrations) + if registration_count: + return True, user_registrations[0] if registration_count == 1 else user_registrations + return False, None + except SQLAlchemyError as e: + self.session.rollback() + raise e def get_unclaimed_reward(self, airdrop_id, address): in_progress_or_completed_tx_statuses = ( @@ -194,9 +192,9 @@ def get_unclaimed_reward(self, airdrop_id, address): def get_registration_by_staking_payment_parts_for_airdrop( self, airdrop_window_id: int, - payment_part: str | None = None, - staking_part: str | None = None, - ) -> UserRegistration | None: + payment_part: Optional[str] = None, + staking_part: Optional[str] = None, + ) -> Optional[UserRegistration]: if not payment_part and not staking_part: raise ValueError("At least one of payment_part / staking_part arguments must be provided") or_clause = list() @@ -211,6 +209,9 @@ def get_registration_by_staking_payment_parts_for_airdrop( UserRegistration.airdrop_window_id == airdrop_window_id, or_(*or_clause) ) + .filter( + UserRegistration.registered_at != None + ) ).first() return registration diff --git a/airdrop/job/eligibility.py b/airdrop/job/eligibility.py index 49698b7..f8cfada 100644 --- a/airdrop/job/eligibility.py +++ b/airdrop/job/eligibility.py @@ -20,7 +20,7 @@ from airdrop.job.repository import Repository from common.exception_handler import exception_handler from common.utils import generate_lambda_response -from airdrop.config import BALANCE_DB_CONFIG, NETWORK, SLACK_HOOK +from airdrop.config import BALANCE_DB_CONFIG, MATTERMOST_CONFIG, NETWORK from common.logger import get_logger from pydoc import locate from decimal import Decimal @@ -141,7 +141,7 @@ def process_eligibility(self): self.__process_reward(processor_name, airdrop_id, window, self._snapshot_guid, False) -@exception_handler(SLACK_HOOK=SLACK_HOOK, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, logger=logger) def process_eligibility(event, context): logger.info(f"Processing eligibility") @@ -158,7 +158,7 @@ def process_eligibility(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, logger=logger) def process_loyalty_airdrop_reward_eligibility(event, context): logger.info(f"Processing loyalty airdrop reward with the event={json.dumps(event)}") @@ -188,7 +188,7 @@ def process_loyalty_airdrop_reward_eligibility(event, context): ) -@exception_handler(SLACK_HOOK=SLACK_HOOK, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, logger=logger) def process_rejuve_airdrop_reward(event, context): # TODO: response format and data (depending on exception_handler) logger.info(f"Processing Rejuve airdrop reward with the event={json.dumps(event)}") @@ -207,7 +207,7 @@ def process_rejuve_airdrop_reward(event, context): # Use for RJV Airdrop ONLY -@exception_handler(SLACK_HOOK=SLACK_HOOK, logger=logger) +@exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, logger=logger) def manual_rejuve_processes(event, context): logger.info(f"Proccesing for the event={json.dumps(event)}") diff --git a/airdrop/job/rejuve_processes.py b/airdrop/job/rejuve_processes.py index 475488b..3ebba7a 100644 --- a/airdrop/job/rejuve_processes.py +++ b/airdrop/job/rejuve_processes.py @@ -5,7 +5,8 @@ from sqlalchemy import or_, select from web3 import Web3 -from airdrop.constants import CARDANO_ADDRESS_PREFIXES, CardanoEra +from airdrop.application.services.common_logic_service import CommonLogicService +from airdrop.constants import CARDANO_ADDRESS_PREFIXES, Blockchain, CardanoEra from airdrop.infrastructure.models import UserBalanceSnapshot, UserRegistration from airdrop.infrastructure.repositories.airdrop_repository import AirdropRepository from airdrop.infrastructure.repositories.user_registration_repo import UserRegistrationRepository @@ -32,8 +33,7 @@ def receive_all_registrations(self) -> list[UserRegistration]: logger.info("Processing the receiving all registrations for the " f"airdrop_id = {self._airdrop_id}, window_id = {self._window_id}") if self.address: - _, registration = UserRegistrationRepository().get_user_registration_details(address=self.address, - airdrop_window_id=self._window_id) + _, registration = CommonLogicService.get_user_registration_details(self.address, self._window_id) registrations = [registration] else: _, registrations = UserRegistrationRepository().get_user_registration_details(airdrop_window_id=self._window_id) @@ -81,8 +81,7 @@ def receive_all_registrations(self) -> list[UserRegistration]: logger.info("Processing the receiving all registrations for the " f"airdrop_id = {self._airdrop_id}, window_id = {self._window_id}") if self.address: - _, registration = UserRegistrationRepository().get_user_registration_details(address=self.address, - airdrop_window_id=self._window_id) + _, registration = CommonLogicService.get_user_registration_details(self.address, self._window_id) registrations = [registration] else: _, registrations = UserRegistrationRepository().get_user_registration_details(airdrop_window_id=self._window_id) @@ -92,19 +91,19 @@ def change_address_format(self, registrations: list[UserRegistration]) -> None: logger.info("Processing the changing address format for the " f"airdrop_id = {self._airdrop_id}, window_id = {self._window_id}") - for addr in registrations: - if (isinstance(addr.address, str) and - Utils().recognize_blockchain_network(addr.address) == "Ethereum"): - user_address = Web3.to_checksum_address(addr.address) - logger.info(f"Old format {addr.address = }. New format {user_address = }") + for registration in registrations: + if (isinstance(registration.address, str) and + Utils.recognize_blockchain_network(registration.address) == Blockchain.ETHEREUM.value): + user_address = Web3.to_checksum_address(registration.address) + logger.info(f"Old format {registration.address = }. New format {user_address = }") UserRegistrationRepository().update_registration_address( airdrop_window_id=self._window_id, - old_address=addr.address, + old_address=registration.address, new_address=user_address ) logger.info("Successfully updated") else: - logger.info(f"Address = {addr.address} is not available for updating") + logger.info(f"Address = {registration.address} is not available for updating") def process_change(self) -> str: if not self._airdrop_id or not self._window_id: diff --git a/airdrop/job/reward_processors/nunet_reward_processor.py b/airdrop/job/reward_processors/nunet_reward_processor.py index df013ad..8bdf794 100644 --- a/airdrop/job/reward_processors/nunet_reward_processor.py +++ b/airdrop/job/reward_processors/nunet_reward_processor.py @@ -2,10 +2,10 @@ import math from decimal import Decimal +from common.alerts import MattermostProcessor from common.exception_handler import exception_handler -from airdrop.config import SLACK_HOOK +from airdrop.config import MATTERMOST_CONFIG from common.logger import get_logger -from common.utils import Utils logger = get_logger(__name__) @@ -87,9 +87,9 @@ def __send_slack_message(self, message, send=False): try: self.__slack_messages.append(message) if send: - u = Utils() + alert_processor = MattermostProcessor(config=MATTERMOST_CONFIG) for message in self.__slack_messages: - u.report_slack(type=0, slack_message=message, slack_config=SLACK_HOOK) + alert_processor.send(type=1, message=message) except Exception as e: logger.info(f"Unable to send slack message {self.__slack_messages}") @@ -105,7 +105,7 @@ def __process_pending_rewards(self): user_pending_rewards[row["address"].lower()] = row["pending_reward"] return additional_rewards, user_pending_rewards - # @exception_handler(SLACK_HOOK=SLACK_HOOK, logger=logger) + # @exception_handler(PROCESSOR_CONFIG=MATTERMOST_CONFIG, logger=logger) def process_rewards(self, only_registered): if only_registered: self.__send_slack_message(f"Computing final rewards for window {self._window_id}") diff --git a/airdrop/job/reward_processors/rejuve_reward_processor.py b/airdrop/job/reward_processors/rejuve_reward_processor.py index 0c258b5..6af7820 100644 --- a/airdrop/job/reward_processors/rejuve_reward_processor.py +++ b/airdrop/job/reward_processors/rejuve_reward_processor.py @@ -143,7 +143,7 @@ def process_user_rewards(self): user_rewards.append(user_reward) self.user_reward_repository.add_all_items(user_rewards) # TODO: batching - def calculate_score(self, balance: Decimal, stake: Decimal) -> (Decimal, Decimal): + def calculate_score(self, balance: Decimal, stake: Decimal) -> tuple[Decimal, Decimal]: score = (balance + self.REWARD_STAKE_RATIO * stake) / self.REWARD_SCORE_DENOM normalized_score = (score + 1).log10() return score, normalized_score diff --git a/airdrop/processor/base_airdrop.py b/airdrop/processor/base_airdrop.py index 2b1e2e2..ed112ee 100644 --- a/airdrop/processor/base_airdrop.py +++ b/airdrop/processor/base_airdrop.py @@ -2,8 +2,11 @@ from datetime import timezone from typing import Tuple, Optional +from web3 import Web3 + from airdrop.config import AIRDROP_RECEIPT_SECRET_KEY, AIRDROP_RECEIPT_SECRET_KEY_STORAGE_REGION -from airdrop.utils import datetime_in_utcnow +from airdrop.constants import Blockchain +from airdrop.utils import Utils, datetime_in_utcnow from common.boto_utils import BotoUtils from common.logger import get_logger from common.utils import get_registration_receipt_ethereum @@ -40,9 +43,12 @@ def is_phase_window_open(start_period, end_period) -> bool: if end_period.tzinfo is None: end_period = end_period.replace(tzinfo=timezone.utc) - if now > start_period and now < end_period: - return True - return False + return start_period <= now <= end_period + + def to_checksum_address_if_ethereum(self, address: str) -> str: + if Utils.recognize_blockchain_network(address) == Blockchain.ETHEREUM.value: + address = Web3.to_checksum_address(address) + return address def generate_user_registration_receipt(self, airdrop_id: int, window_id: int, address: str) -> str: diff --git a/airdrop/processor/rejuve_airdrop.py b/airdrop/processor/rejuve_airdrop.py index 87d09dc..ff4ba60 100644 --- a/airdrop/processor/rejuve_airdrop.py +++ b/airdrop/processor/rejuve_airdrop.py @@ -8,7 +8,8 @@ from pycardano import Address from web3 import Web3 -from airdrop.constants import CARDANO_ADDRESS_PREFIXES, AirdropClaimStatus, CardanoEra, TransactionType +from airdrop.application.services.common_logic_service import CommonLogicService +from airdrop.constants import CARDANO_ADDRESS_PREFIXES, AirdropClaimStatus, Blockchain, CardanoEra, TransactionType from airdrop.infrastructure.models import AirdropWindow, UserRegistration from airdrop.application.types.windows import WindowRegistrationData from airdrop.infrastructure.repositories.airdrop_repository import AirdropRepository @@ -42,9 +43,10 @@ def __init__(self, airdrop_id, airdrop_window_id=None): self.claim_address = RejuveAirdropConfig.claim_address.value def check_user_eligibility(self, address: str) -> bool: - if any(address.startswith(prefix) for prefix in CARDANO_ADDRESS_PREFIXES[CardanoEra.SHELLEY]): + if address.startswith(tuple(CARDANO_ADDRESS_PREFIXES[CardanoEra.SHELLEY])): formatted_address = Address.from_primitive(address) + # TODO (Potential Error): Cardano address without staking part won't pass it if formatted_address.payment_part is not None and formatted_address.staking_part is not None: balances = UserBalanceSnapshotRepository().get_balances_by_staking_payment_parts_for_airdrop( payment_part=str(formatted_address.payment_part), @@ -52,6 +54,7 @@ def check_user_eligibility(self, address: str) -> bool: airdrop_id=self.id ) + # TODO: Zero balance checker for existing rows return bool(balances and len(balances) > 0) logger.error(f"Staking and payment part not found for address: {address}") @@ -108,12 +111,11 @@ def match_signature( network = Utils.recognize_blockchain_network(address) logger.info(f"Start of signature matching | address={address}, signature={signature}, network={network}") - if network not in {"Ethereum", "Cardano"}: + if network not in {Blockchain.ETHEREUM.value, Blockchain.CARDANO.value}: raise ValueError(f"Unsupported network: {network}") - if network == "Ethereum": - address = Web3.to_checksum_address(address) - elif network == "Cardano" and key is None: + address = self.to_checksum_address_if_ethereum(address) + if network == Blockchain.CARDANO.value and key is None: raise ValueError("Key must be provided for Cardano signatures.") formatted_message = self.format_user_registration_signature_message( @@ -125,7 +127,7 @@ def match_signature( sign_verified = ( Utils.match_ethereum_signature_eip191(address, message, signature) - if network == "Ethereum" + if network == Blockchain.ETHEREUM.value else Utils.match_cardano_signature(message, signature, key) # type: ignore ) @@ -147,8 +149,7 @@ def generate_user_registration_receipt( raise Exception("Window ID is not set") try: - if Utils.recognize_blockchain_network(address) == "Ethereum": - address = Web3.to_checksum_address(address) + address = self.to_checksum_address_if_ethereum(address) message = Web3.solidity_keccak( ["string", "string", "uint256", "uint256", "uint256"], @@ -202,8 +203,7 @@ def register_regular_wallet(self, data: dict) -> list | str: if self.window_id is None: raise Exception("Window ID is None") - if Utils.recognize_blockchain_network(address) == "Ethereum": - address = Web3.to_checksum_address(address) + address = self.to_checksum_address_if_ethereum(address) registration_repo = UserRegistrationRepository() airdrop_window_repo = AirdropWindowRepository() @@ -233,14 +233,14 @@ def register_regular_wallet(self, data: dict) -> list | str: logger.error("Address is not eligible for this airdrop") raise Exception("Address is not eligible for this airdrop") - is_registered, _ = registration_repo.get_user_registration_details(address, self.window_id) + is_registered, _ = CommonLogicService.get_user_registration_details(address, self.window_id) if is_registered: logger.error("Address is already registered for this airdrop window") raise Exception("Address is already registered for this airdrop window") payment_part: str | None = None staking_part: str | None = None - if any(address.startswith(prefix) for prefix in CARDANO_ADDRESS_PREFIXES[CardanoEra.SHELLEY]): + if address.startswith(tuple(CARDANO_ADDRESS_PREFIXES[CardanoEra.SHELLEY])): formatted_address = Address.from_primitive(address) payment_part = str(formatted_address.payment_part) if formatted_address.payment_part else None staking_part = str(formatted_address.staking_part) if formatted_address.staking_part else None @@ -277,8 +277,7 @@ def register_trezor(self, data: dict) -> list | str: if self.window_id is None: raise Exception("Window ID is None") - if Utils.recognize_blockchain_network(address) == "Ethereum": - address = Web3.to_checksum_address(address) + address = self.to_checksum_address_if_ethereum(address) registration_repo = UserRegistrationRepository() pending_registration_repo = PendingTransactionRepository() @@ -302,14 +301,14 @@ def register_trezor(self, data: dict) -> list | str: logger.error("Address is not eligible for this airdrop") raise Exception("Address is not eligible for this airdrop") - is_registered, _ = registration_repo.get_user_registration_details(address, self.window_id) + is_registered, _ = CommonLogicService.get_user_registration_details(address, self.window_id) if is_registered: logger.error("Address is already registered for this airdrop window") raise Exception("Address is already registered for this airdrop window") payment_part: str | None = None staking_part: str | None = None - if any(address.startswith(prefix) for prefix in CARDANO_ADDRESS_PREFIXES[CardanoEra.SHELLEY]): + if address.startswith(tuple(CARDANO_ADDRESS_PREFIXES[CardanoEra.SHELLEY])): formatted_address = Address.from_primitive(address) payment_part = str(formatted_address.payment_part) if formatted_address.payment_part else None staking_part = str(formatted_address.staking_part) if formatted_address.staking_part else None @@ -347,155 +346,134 @@ def register_trezor(self, data: dict) -> list | str: return receipt def update_registration(self, data: dict): - logger.info(f"Starting registration update process for {self.__class__.__name__}") - - if "tx_hash" in data: - return self.update_registration_trezor(data) - else: - return self.update_registration_regular_wallet(data) - - def update_registration_regular_wallet(self, data: dict): """ Update the user's registration details for a specific airdrop window. Steps: - 1. Validate the airdrop window exists. - 2. Check if claim phase is open. - 3. Check if the user is eligible for the airdrop. - 4. Match the provided signature to confirm identity. - 5. Ensure the user is already registered. - 6. Generate new receipt. - 7. Update the registration with new signature details. + 1. Check whether airdrop claim phase is open + 2. Check whether the user is eligible for the airdrop + 3. Select registered & claimable airdrop window for update + 3.1 Validate that requested airdrop window exists + 3.2 Validate that there is claimable window for the user in this airdrop + 4. Validate signature data (4.1 OR 4.2) + 4.1 Validate transaction for trezor wallet type + 4.2 Match the provided signature to confirm identity + 5. Generate new receipt + 6. Update the registration with new signature details + 7. Calculate current claimable amount for the user """ - logger.info(f"Starting regular wallet registration update process for {self.__class__.__name__}") + logger.info(f"Starting registration update process for {self.__class__.__name__}") address = data["address"] - signature = data["signature"] reward_address = data["reward_address"] timestamp = data["timestamp"] wallet_name = data["wallet_name"] + signature = data.get("signature") key = data.get("key") - - if self.window_id is None: - raise Exception("Window ID is None") + tx_hash = data.get("tx_hash") registration_repo = UserRegistrationRepository() airdrop_window_repo = AirdropWindowRepository() + claim_history_repo = ClaimHistoryRepository() - if Utils.recognize_blockchain_network(address) == "Ethereum": - address = Web3.to_checksum_address(address) - - airdrop_window: AirdropWindow = airdrop_window_repo.get_airdrop_window_by_id(self.window_id) - if not airdrop_window: - raise Exception(f"Airdrop window does not exist: {self.window_id}") + address = self.to_checksum_address_if_ethereum(address) - if not self.is_phase_window_open( - airdrop_window.claim_start_period, - airdrop_window.claim_end_period - ): - raise Exception("Airdrop window is not accepting claim at this moment") + # Check whether airdrop claim phase is open + now = datetime_in_utcnow() + if not airdrop_window_repo.is_claimable_airdrop(airdrop_id=self.id, date_time=now): + raise Exception("Claim is not available at this moment") + # Check whether the user is eligible for the airdrop if not self.check_user_eligibility(address=address): - logger.error(f"Address {address} is not eligible for airdrop in window {self.window_id}") - raise Exception("Address is not eligible for this airdrop.") - - claim_history_obj = ClaimHistoryRepository().get_claim_history(self.window_id, address, "ada_transfer") - if claim_history_obj: - logger.error(f"Claim transaction already created for {address = }, " - f"window_id = {self.window_id} and blockchain_method = ada_transfer") - raise Exception("It is forbidden to update the data because a claim " - "transaction has already been created for it") + logger.error(f"Address {address} is not eligible for airdrop {self.id}") + raise Exception("Address is not eligible for this airdrop") - signature_details = self.match_signature( - address=address, - signature=signature, - timestamp=timestamp, - wallet_name=wallet_name, - key=key, - reward_address=reward_address - ) + # Select registered & claimable airdrop window for update + requested_airdrop_window: AirdropWindow = airdrop_window_repo.get_airdrop_window_by_id(self.window_id) + if not requested_airdrop_window: + raise Exception(f"Airdrop window does not exist: {requested_airdrop_window.id}") + claimable_airdrop_window = None + airdrop_windows = airdrop_window_repo.get_airdrop_windows(self.id) + for airdrop_window_ in airdrop_windows: + is_registered, _ = CommonLogicService.get_user_registration_details(address, airdrop_window_.id) + claim_history_obj = claim_history_repo.get_claim_history(airdrop_window_.id, address, "ada_transfer") + is_claimed = claim_history_obj is not None + is_after_requested = airdrop_window_.airdrop_window_order > requested_airdrop_window.airdrop_window_order + logger.debug(f"Airdrop window id={airdrop_window_.id} is_registered={is_registered}, " + f"is_claimed={is_claimed}, is_after_requested={is_after_requested}") + if is_registered and not is_claimed and not is_after_requested: + claimable_airdrop_window = airdrop_window_ + elif is_registered and is_claimed: + claimable_airdrop_window = None + if isinstance(claimable_airdrop_window, AirdropWindow): + logger.info(f"Selected claimable window id={claimable_airdrop_window.id}") + else: + raise Exception(f"Claimable window not found for the requested window: {requested_airdrop_window.id}") - is_registered, _ = registration_repo.get_user_registration_details(address, self.window_id) - if not is_registered: - logger.error(f"Address {address} is not registered for window {self.window_id}") - raise Exception("Address is not registered for this airdrop window.") + if tx_hash: + # Validate transaction for trezor wallet type + signature_details = self.validate_update_registration_trezor( + window_id=claimable_airdrop_window.id, + address=address, + reward_address=reward_address, + timestamp=timestamp, + wallet_name=wallet_name, + tx_hash=tx_hash + ) + elif signature: + # Match the provided signature to confirm identity + signature_details = self.match_signature( + address=address, + signature=signature, + timestamp=timestamp, + wallet_name=wallet_name, + key=key, + reward_address=reward_address + ) + else: + raise Exception("Validation data not provided (required either a signature or a tx_hash)") + # Generate new receipt receipt = self.get_receipt(address=address, timestamp=timestamp) + # Update the registration with new signature details registration_repo.update_registration( - airdrop_window_id=self.window_id, + airdrop_window_id=claimable_airdrop_window.id, address=address, - signature=signature, + signature=signature if signature and not tx_hash else None, signature_details=signature_details, + tx_hash=tx_hash if tx_hash else None, receipt=receipt ) + # Calculate current claimable amount for the user claimable_amount, total_eligible_amount = self.get_claimable_amount(user_address=address) return { "airdrop_id": str(self.id), - "airdrop_window_id": str(airdrop_window.id), + "airdrop_window_id": str(claimable_airdrop_window.id), "claimable_amount": str(claimable_amount), "total_eligibility_amount": str(total_eligible_amount), "chain_context": self.chain_context, "registration_id": receipt } - def update_registration_trezor(self, data: dict): + def validate_update_registration_trezor(self, window_id, address, reward_address, timestamp, wallet_name, tx_hash): """ - Update the user's registration details for a specific airdrop window. + Validate update registration process for trezor wallet type. Steps: - 1. Validate the airdrop window exists. - 2. Check if claim phase is open. - 3. Check if the user is eligible for the airdrop. - 4. Ensure the user is already registered. - 5. Transaction address, transaction metadata and timestamp checks - 6. Generate new receipt. - 7. Update the registration with new signature details. + 1. Transaction address check + 2. Transaction metadata check + 3. Transaction timestamp from metadata check """ - logger.info(f"Starting trezor wallet registration update process for {self.__class__.__name__}") - - address = data["address"] - reward_address = data["reward_address"] - timestamp = data["timestamp"] - wallet_name = data["wallet_name"] - tx_hash = data["tx_hash"] - - if self.window_id is None: - raise Exception("Window ID is None") - - registration_repo = UserRegistrationRepository() - airdrop_window_repo = AirdropWindowRepository() - - if Utils.recognize_blockchain_network(address) == "Ethereum": - address = Web3.to_checksum_address(address) - - airdrop_window: AirdropWindow = airdrop_window_repo.get_airdrop_window_by_id(self.window_id) - if not airdrop_window: - raise Exception(f"Airdrop window does not exist: {self.window_id}") - - if not self.is_phase_window_open( - airdrop_window.claim_start_period, - airdrop_window.claim_end_period - ): - raise Exception("Airdrop window is not accepting claim at this moment") - - if not self.check_user_eligibility(address=address): - logger.error(f"Address {address} is not eligible for airdrop in window {self.window_id}") - raise Exception("Address is not eligible for this airdrop.") + logger.info(f"Starting validation for trezor wallet registration update process") - claim_history_obj = ClaimHistoryRepository().get_claim_history(self.window_id, address, "ada_transfer") - if claim_history_obj: - logger.error(f"Claim transaction already created for {address = }, " - f"window_id = {self.window_id} and blockchain_method = ada_transfer") - raise Exception("It is forbidden to update the data because a claim " - "transaction has already been created for it") - - is_registered, registration = registration_repo.get_user_registration_details(address, self.window_id) - if not is_registered: - logger.error(f"Address {address} is not registered for window {self.window_id}") - raise Exception("Address is not registered for this airdrop window.") + signature_details = self.format_trezor_user_registration_signature_message( + timestamp=timestamp, + wallet_name=wallet_name, + ) + signature_details["walletAddress"] = reward_address.lower() blockfrost = BlockFrostApi(project_id=BlockFrostAccountDetails.project_id, base_url=BlockFrostAPIBaseURL) @@ -506,7 +484,7 @@ def update_registration_trezor(self, data: dict): logger.info(f"Found tx {tx_hash}: block={tx_data.block_height} index={tx_data.index}") except BlockFrostApiError as error: logger.exception(f"BlockFrostApiError: {error}") - raise TransactionNotFound(f"Transaction with {tx_hash = } not found in blockchain") + raise TransactionNotFound(f"Transaction with {tx_hash=} not found in the blockchain") # Transaction address check is_address_match = False @@ -517,46 +495,23 @@ def update_registration_trezor(self, data: dict): if not is_address_match: raise Exception("Transaction address is not valid") - formatted_message = self.format_trezor_user_registration_signature_message( - timestamp=timestamp, - wallet_name=wallet_name, - ) - # Metadata check - is_metadata_match, metadata = Utils().compare_data_from_db_and_metadata( - formatted_message, + # Transaction metadata check + is_metadata_match, metadata = Utils.compare_data_from_db_and_metadata( + signature_details, tx_metadata ) if not is_metadata_match: raise Exception("Transaction metadata is not valid") - # Timestamp from metadata check + # Transaction timestamp from metadata check is_transaction_newest = False + _, registration = CommonLogicService.get_user_registration_details(address, window_id) if metadata["timestamp"] > registration.signature_details["timestamp"]: is_transaction_newest = True if not is_transaction_newest: raise Exception("The transaction you passed is older than the one you used previously") - formatted_message["walletAddress"] = reward_address.lower() - receipt = self.get_receipt(address=address, timestamp=timestamp) - - registration_repo.update_registration( - airdrop_window_id=self.window_id, - address=address, - signature_details=formatted_message, - tx_hash=tx_hash, - receipt=receipt - ) - - claimable_amount, total_eligible_amount = self.get_claimable_amount(user_address=address) - - return { - "airdrop_id": str(self.id), - "airdrop_window_id": str(airdrop_window.id), - "claimable_amount": str(claimable_amount), - "total_eligibility_amount": str(total_eligible_amount), - "chain_context": self.chain_context, - "registration_id": receipt - } + return signature_details def generate_multiple_windows_eligibility_response( self, diff --git a/airdrop/utils.py b/airdrop/utils.py index 89ecc03..9d14709 100644 --- a/airdrop/utils.py +++ b/airdrop/utils.py @@ -5,11 +5,13 @@ import cbor2 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from eth_account.messages import encode_defunct, encode_typed_data +from pycardano import Address import requests from web3 import Web3 from airdrop.config import BlockFrostAccountDetails -from airdrop.constants import BlockFrostAPI +from airdrop.constants import (CARDANO_ADDRESS_PREFIXES, BlockFrostAPI, + Blockchain, CardanoEra) from common.logger import get_logger logger = get_logger(__name__) @@ -22,10 +24,10 @@ def datetime_in_utcnow(): class Utils: @staticmethod def recognize_blockchain_network(address: str) -> str: - if address[:2] == "0x": - return "Ethereum" - elif address[:4] == "addr": - return "Cardano" + if address.startswith("0x"): + return Blockchain.ETHEREUM.value + elif address.startswith(tuple(CARDANO_ADDRESS_PREFIXES[CardanoEra.ANY])): + return Blockchain.CARDANO.value else: return "Unknown" @@ -128,4 +130,14 @@ def compare_data_from_db_and_metadata( tx_value = metadata[0].json_metadata.to_dict() if registration_message == tx_value: return True, tx_value - return False, None \ No newline at end of file + return False, None + + @staticmethod + def get_payment_staking_parts(address: str) -> tuple[str | None, str | None]: + payment_part: str | None = None + staking_part: str | None = None + if address.startswith(tuple(CARDANO_ADDRESS_PREFIXES[CardanoEra.SHELLEY])): + formatted_address = Address.from_primitive(address) + payment_part = str(formatted_address.payment_part) if formatted_address.payment_part else None + staking_part = str(formatted_address.staking_part) if formatted_address.staking_part else None + return payment_part, staking_part diff --git a/common/alerts.py b/common/alerts.py new file mode 100644 index 0000000..0e5bae3 --- /dev/null +++ b/common/alerts.py @@ -0,0 +1,134 @@ +import json +import requests +from abc import ABC, abstractmethod +from functools import wraps +from common.constant import ERROR_MESSAGE_FORMAT +from common.logger import get_logger + +logger = get_logger(__name__) + + +def delivery_exception_handler(method): + + @wraps(method) + def wrapper(self, *args, **kwargs): + try: + self.health_check() + return method(self, *args, **kwargs) + except AssertionError as e: + logger.warning(f"Failed to send alert ({e})", exc_info=False) + except Exception as e: + logger.warning(f"Failed to send alert ({repr(e)})", exc_info=True) + + return wrapper + + +class AlertsProcessor(ABC): + + @property + @abstractmethod + def name(self): + """Set the name of the alerts processor for external logging and debug""" + return "Abstract" + + @abstractmethod + def __init__(self): + """Definition of all the parameters required to deliver notifications to the alert channel""" + pass + + @abstractmethod + def health_check(self): + """Check that all processor parameters are set correctly for successful delivery + !!! For each check use only assertions without AssertionError handling + """ + assert True, "Check not passed" # Use constructions like this to check all important points + + @abstractmethod + def prepare_error_message(self, type: int, message: str) -> str: + pass + + @abstractmethod + def send(self, message: str): + """Main method to deliver notification message to the alert channel""" + pass + + +class SlackProcessor(AlertsProcessor): + + @property + def name(self): + return "Slack" + + def __init__(self, config: dict) -> None: + self.msg_type = {0: "Info:: ", 1: "Err:: "} + self.url = config.get("hostname") + config.get("path") + self.channel = config.get("channel_name") + self.username = "webhookbot" + self.icon = ":ghost:" + + def health_check(self) -> None: + assert self.url and isinstance(self.url, str), "Bad slack url value" + assert self.channel and isinstance(self.channel, str), "Bad slack channel value" + + def prepare_error_message(self, network_id, query_string_parameters, + path, handler_name, path_parameters, body, + error_description) -> str: + message = ERROR_MESSAGE_FORMAT.format( + network_id=network_id, + query_string_parameters=query_string_parameters, + path=path, + handler_name=handler_name, + path_parameters=path_parameters, + body=body, + error_description=error_description + ) + return f"```{message}```" + + @delivery_exception_handler + def send(self, type: int, message: str) -> None: + logger.info(f"Sending slack notification to #{self.channel}") + prefix = self.msg_type.get(type, "") + payload = { + "channel": f"#{self.channel}", + "username": self.username, + "icon_emoji": self.icon, + "text": prefix + message + } + response = requests.post(url=self.url, data=json.dumps(payload)) + logger.info(f"{self.name} response [code {response.status_code}]: {response.text}") + + +class MattermostProcessor(AlertsProcessor): + + @property + def name(self): + return "Mattermost" + + def __init__(self, config: dict) -> None: + self.msg_type = {0: ":information_source: ", 1: ":warning: "} + self.url = config.get("url") + + def health_check(self) -> None: + assert self.url and isinstance(self.url, str), "Bad mattermost url value" + + def prepare_error_message(self, network_id, query_string_parameters, + path, handler_name, path_parameters, body, + error_description) -> str: + message = ERROR_MESSAGE_FORMAT.format( + network_id=network_id, + query_string_parameters=query_string_parameters, + path=path, + handler_name=handler_name, + path_parameters=path_parameters, + body=body, + error_description=error_description + ) + return message + + @delivery_exception_handler + def send(self, type: int, message: str) -> None: + headers = {"Content-Type": "application/json"} + prefix = "### " + self.msg_type.get(type, "") + payload = {"text": prefix + message} + response = requests.post(self.url, headers=headers, data=json.dumps(payload)) + logger.info(f"{self.name} response [code {response.status_code}]: {response.text}") diff --git a/common/constant.py b/common/constant.py index ff7565c..71a3f28 100644 --- a/common/constant.py +++ b/common/constant.py @@ -10,7 +10,7 @@ class ResponseStatus: SUCCESS = "success" -SLACK_ERROR_MESSAGE_FORMAT = ( +ERROR_MESSAGE_FORMAT = ( "Error Reported! \n" "network_id: {network_id}\n" "path: {path}, \n" diff --git a/common/exception_handler.py b/common/exception_handler.py index f3834c2..67e4283 100644 --- a/common/exception_handler.py +++ b/common/exception_handler.py @@ -1,9 +1,11 @@ -import json +from logging import Logger import sys import traceback -from common.constant import SLACK_ERROR_MESSAGE_FORMAT, ResponseStatus, StatusCode -from common.utils import Utils, generate_lambda_response +from common.alerts import MattermostProcessor +from common.constant import ResponseStatus, StatusCode +from common.exceptions import BlockConfirmationException +from common.utils import generate_lambda_response def extract_parameters_from_event(event): @@ -14,14 +16,6 @@ def extract_parameters_from_event(event): return path, path_parameters, query_string_parameters, body -def prepare_slack_error_message(network_id, query_string_parameters, path, handler_name, path_parameters, body, - error_description): - message = SLACK_ERROR_MESSAGE_FORMAT.format(network_id=network_id, query_string_parameters=query_string_parameters, - path=path, handler_name=handler_name, path_parameters=path_parameters, - body=body, error_description=error_description) - return message - - def get_error_description(e): exc_type, exc_obj, exc_tb = sys.exc_info() exc_tb_lines = traceback.format_tb(exc_tb) @@ -44,44 +38,52 @@ def prepare_response_message(status, data="", code=0, error_message="", error_de def exception_handler(*decorator_args, **decorator_kwargs): - logger = decorator_kwargs["logger"] - network_id = decorator_kwargs.get("NETWORK_ID", None) - slack_hook = decorator_kwargs.get("SLACK_HOOK", None) - exceptions = decorator_kwargs.get("EXCEPTIONS", ()) - raise_exception = decorator_kwargs.get("RAISE_EXCEPTION", False) + logger: Logger = decorator_kwargs["logger"] + network_id: int | None = decorator_kwargs.get("NETWORK_ID", None) + processor_config: dict = decorator_kwargs.get("PROCESSOR_CONFIG", {}) + not_raised_exceptions: tuple = decorator_kwargs.get("NOT_RAISED_EXCEPTIONS", ()) + raised_exceptions: tuple = decorator_kwargs.get("RAISED_EXCEPTIONS", ()) + raise_exception: bool = decorator_kwargs.get("RAISE_EXCEPTION", False) + + alert_processor = MattermostProcessor(config=processor_config) def decorator(func): def wrapper(*args, **kwargs): handler_name = decorator_kwargs.get("handler_name", func.__name__) - event = kwargs.get("event", args[0] if len(args) >0 else {}) + event = kwargs.get("event", args[0] if len(args) > 0 else {}) path, path_parameters, query_string_parameters, body = extract_parameters_from_event(event) error_description = "" try: return func(*args, **kwargs) - except exceptions as e: + except not_raised_exceptions as e: error_description = get_error_description(e) - slack_error_message = prepare_slack_error_message(network_id=network_id, - query_string_parameters=query_string_parameters, - path=path, handler_name=handler_name, - path_parameters=path_parameters, - body=body, error_description=error_description) - slack_message = f"```{slack_error_message}```" - logger.exception(slack_error_message) - Utils().report_slack(type=0, slack_message=slack_message, slack_config=slack_hook) + error_message = alert_processor.prepare_error_message(network_id=network_id, + query_string_parameters=query_string_parameters, + path=path, + handler_name=handler_name, + path_parameters=path_parameters, + body=body, + error_description=error_description) + logger.exception(error_description) + alert_processor.send(type=1, message=error_message) response_message = prepare_response_message(ResponseStatus.FAILED, error_message=e.error_message, error_details=e.error_details) return generate_lambda_response(StatusCode.INTERNAL_SERVER_ERROR, response_message, cors_enabled=True) + except raised_exceptions as e: + error_description = get_error_description(e) + logger.exception(error_description) + raise e except Exception as e: error_description = get_error_description(e) - slack_error_message = prepare_slack_error_message(network_id=network_id, - query_string_parameters=query_string_parameters, - path=path, handler_name=handler_name, - path_parameters=path_parameters, - body=body, error_description=error_description) - logger.exception(slack_error_message) - slack_message = f"```{slack_error_message}```" - logger.exception(slack_error_message) - Utils().report_slack(type=0, slack_message=slack_message, slack_config=slack_hook) + error_message = alert_processor.prepare_error_message(network_id=network_id, + query_string_parameters=query_string_parameters, + path=path, + handler_name=handler_name, + path_parameters=path_parameters, + body=body, + error_description=error_description) + logger.exception(error_description) + alert_processor.send(type=1, message=error_message) response_message = prepare_response_message(ResponseStatus.FAILED, error_details=repr(e)) if raise_exception: raise e diff --git a/common/exceptions.py b/common/exceptions.py index 2edf282..dca5274 100644 --- a/common/exceptions.py +++ b/common/exceptions.py @@ -22,3 +22,7 @@ def __init__(self, error_details): class TransactionNotFound(Exception): pass + + +class BlockConfirmationException(Exception): + pass diff --git a/common/utils.py b/common/utils.py index 5289b56..0dca493 100644 --- a/common/utils.py +++ b/common/utils.py @@ -18,26 +18,6 @@ class ContractType(Enum): AIRDROP = "AIRDROP" -class Utils: - def __init__(self): - self.msg_type = {0: "Info:: ", 1: "Err:: "} - - def report_slack(self, type, slack_message, slack_config): - url = slack_config["hostname"] + slack_config["path"] - prefix = self.msg_type.get(type, "") - slack_channel = slack_config.get("channel", SLACK_HOOK['channel_name']) - print(url) - payload = { - "channel": f"#{slack_channel}", - "username": "webhookbot", - "text": prefix + slack_message, - "icon_emoji": ":ghost:", - } - - resp = requests.post(url=url, data=json.dumps(payload)) - print(resp.status_code, resp.text) - - def request(event) -> dict | None: try: inputs = event["body"] or None