free upload & player

app/core/background/ton_service.py

@@ -1,8 +1,11 @@
 import asyncio
+from base64 import b64decode
 import os
+import traceback
+import httpx
 
-from sqlalchemy import and_
-from tonsdk.boc import begin_cell
+from sqlalchemy import and_, func
+from tonsdk.boc import begin_cell, Cell
 from tonsdk.contract.wallet import Wallets
 from tonsdk.utils import HighloadQueryId
 from datetime import datetime, timedelta
@@ -93,7 +96,7 @@ async def main_fn(memory):
         await toncenter.send_boc(
             service_wallet.create_transfer_message(
                 [{
-                    'address': highload_wallet.address.to_string(1, 1, 1),
+                    'address': highload_wallet.address.to_string(1, 1, 0),
                     'amount': int(0.08 * 10 ** 9),
                     'send_mode': 1,
                     'payload': begin_cell().store_uint(0, 32).end_cell()
@@ -122,18 +125,153 @@ async def main_fn(memory):
             sw_seqno_value = await get_sw_seqno()
             make_log("TON", f"Service running ({sw_seqno_value})", level="debug")
 
-            # with db_session() as session:
-            #     next_task = session.query(BlockchainTask).filter(
-            #         BlockchainTask.status == 'wait'
-            #     ).first()
-            #     if next_task:
-            #         make_log("TON", f"Processing task {next_task.id}", level="info")
-                    
-            #         next_task.status = 'processing'
+            with db_session() as session:
+                # Проверка отправленных сообщений
+                async def process_incoming_transaction(transaction: dict):
+                    transaction_hash = transaction['transaction_id']['hash']
+                    transaction_lt = str(transaction['transaction_id']['lt'])
+                    # transaction_success = bool(transaction['success'])
 
-            #         session.commit()
+                    async def process_incoming_message(blockchain_message: dict):
+                        in_msg_cell = Cell.one_from_boc(b64decode(blockchain_message['msg_data']['body']))
+                        in_msg_slice = in_msg_cell.refs[0].begin_parse()
+                        in_msg_slice.read_uint(32)
+                        in_msg_slice.read_uint(8)
+                        in_msg_query_id = in_msg_slice.read_uint(23)
+                        in_msg_created_at = in_msg_slice.read_uint(64)
+                        in_msg_epoch = int(in_msg_created_at // (60 * 60))
+                        in_msg_seqno = HighloadQueryId.from_query_id(in_msg_query_id).to_seqno()
+                        in_msg_blockchain_task = (
+                            session.query(BlockchainTask).filter(
+                                and_(
+                                    BlockchainTask.seqno == in_msg_seqno,
+                                    BlockchainTask.epoch == in_msg_epoch,
+                                )
+                            )
+                        ).first()
+                        if not in_msg_blockchain_task:
+                            return
 
-            await asyncio.sleep(5)
+                        if not (in_msg_blockchain_task.status in ['done']) or in_msg_blockchain_task.transaction_hash != transaction_hash:
+                            in_msg_blockchain_task.status = 'done'
+                            in_msg_blockchain_task.transaction_hash = transaction_hash
+                            in_msg_blockchain_task.transaction_lt = transaction_lt
+                            await session.commit()
+
+                    for blockchain_message in [transaction['in_msg']]:
+                        try:
+                            await process_incoming_message(blockchain_message)
+                        except BaseException as e:
+                            pass # make_log("TON_Daemon", f"Error while processing incoming message: {e}" + '\n' + traceback.format_exc(), level='debug')
+
+                try:
+                    sw_transactions = await toncenter.get_transactions(highload_wallet.address.to_string(1, 1, 1), limit=100)
+                    for sw_transaction in sw_transactions:
+                        try:
+                            await process_incoming_transaction(sw_transaction)
+                        except BaseException as e:
+                            make_log("TON_Daemon", f"Error while processing incoming transaction: {e}", level="debug")
+                except BaseException as e:
+                    make_log("TON_Daemon", f"Error while getting service wallet transactions: {e}", level="ERROR")
+
+                # Отправка подписанных сообщений
+                for blockchain_task in (
+                    session.query(BlockchainTask).filter(
+                        BlockchainTask.status == 'processing',
+                    ).order_by(BlockchainTask.updated.asc()).all()
+                ):
+                    make_log("TON_Daemon", f"Processing task (processing) {blockchain_task.id}")
+                    query_boc = bytes.fromhex(blockchain_task.meta['signed_message'])
+                    errors_list = []
+
+                    try:
+                        await toncenter.send_boc(query_boc)
+                    except BaseException as e:
+                        errors_list.append(f"{e}")
+
+                    try:
+                        make_log("TON_Daemon", str(
+                            httpx.post(
+                                'https://tonapi.io/v2/blockchain/message',
+                                json={
+                                    'boc': query_boc.hex()
+                                }
+                            ).text
+                        ))
+                    except BaseException as e:
+                        make_log("TON_Daemon", f"Error while pushing task to tonkeeper ({blockchain_task.id}): {e}", level="ERROR")
+                        errors_list.append(f"{e}")
+
+                    blockchain_task.updated = datetime.utcnow()
+
+                    if blockchain_task.meta['sign_created'] + 10 * 60 < datetime.utcnow().timestamp():
+                            # or sum([int("terminating vm with exit code 36" in e) for e in errors_list]) > 0:
+                        make_log("TON_Daemon", f"Task {blockchain_task.id} done", level="DEBUG")
+                        blockchain_task.status = 'done'
+                        await session.commit()
+                        continue
+
+                    await asyncio.sleep(0.5)
+
+                # Создание новых подписей
+                for blockchain_task in (
+                    session.query(BlockchainTask).filter(BlockchainTask.status == 'wait').all()
+                ):
+                    try:
+                        # Check processing tasks in current epoch < 3_000_000
+                        if (
+                            session.query(BlockchainTask).filter(
+                                BlockchainTask.epoch == blockchain_task.epoch,
+                            ).count() > 3_000_000
+                        ):
+                            make_log("TON", f"Too many processing tasks in epoch {blockchain_task.epoch}", level="error")
+                            await send_status("ton_daemon", f"working: too many tasks in epoch {blockchain_task.epoch}")
+                            await asyncio.sleep(5)
+                            continue
+
+                        sign_created = int(datetime.utcnow().timestamp()) - 60
+                        try:
+                            current_epoch = int(datetime.utcnow().timestamp() // (60 * 60))
+                            max_epoch_seqno = (
+                                session.query(func.max(BlockchainTask.seqno)).filter(
+                                    BlockchainTask.epoch == current_epoch
+                                ).scalar() or 0
+                            )
+                            current_epoch_shift = 3_000_000 if current_epoch % 2 == 0 else 0
+                            current_seqno = max_epoch_seqno + 1 + (current_epoch_shift if max_epoch_seqno == 0 else 0)
+                        except BaseException as e:
+                            make_log("CRITICAL", f"Error calculating epoch,seqno: {e}", level="error")
+                            current_epoch = 0
+                            current_seqno = 0
+
+                        blockchain_task.seqno = current_seqno
+                        blockchain_task.epoch = current_epoch
+                        blockchain_task.status = 'processing'
+                        try:
+                            query = highload_wallet.create_transfer_message(
+                                blockchain_task.destination, int(blockchain_task.amount), HighloadQueryId.from_seqno(current_seqno),
+                                sign_created, send_mode=1,
+                                payload=Cell.one_from_boc(b64decode(blockchain_task.payload))
+                            )
+                            query_boc = query['message'].to_boc(False)
+                        except BaseException as e:
+                            make_log("TON", f"Error creating transfer message: {e}", level="error")
+                            query_boc = begin_cell().end_cell().to_boc(False)
+                        
+                        blockchain_task.meta = {
+                            **blockchain_task.meta,
+                            'sign_created': sign_created,
+                            'signed_message': query_boc,
+                        }
+                        await session.commit()
+                        make_log("TON", f"Created signed message for task {blockchain_task.id}" + '\n' + traceback.format_exc(), level="info")
+                    except BaseException as e:
+                        make_log("TON", f"Error processing task {blockchain_task.id}: {e}" + '\n' + traceback.format_exc(), level="error")
+                        continue
+
+                    await asyncio.sleep(1)
+
+            await asyncio.sleep(1)
             await send_status("ton_daemon", f"working (seqno={sw_seqno_value})")
         except BaseException as e:
             make_log("TON", f"Error: {e}", level="error")

app/core/models/_telegram/templates/player.py

@@ -182,7 +182,7 @@ class PlayerTemplates:
                         )
                 ).first())
             )
-            if have_access:
+            if False and have_access:
                 full_content = self.db_session.query(StoredContent).filter_by(
                     hash=content.meta.get('converted_content', {}).get('low')  # TODO: support high quality
                 ).first()

app/core/models/tasks.py

@@ -9,7 +9,8 @@ class BlockchainTask:
     id = Column(Integer, autoincrement=True, primary_key=True)
     
     destination = Column(String(1024), nullable=False)
-    payload = Column(String(1024), nullable=False)
+    amount = Column(String(256), nullable=False)
+    payload = Column(String(4096), nullable=False)
 
     epoch = Column(Integer, nullable=True)
     seqno = Column(Integer, nullable=True)
@@ -17,6 +18,7 @@ class BlockchainTask:
     created = Column(DateTime, nullable=False, default=datetime.now)
     updated = Column(DateTime, nullable=False, default=datetime.now)
 
+    user_id = Column(Integer, ForeignKey('users.id'), nullable=True)
     meta = Column(JSON, nullable=False, default={})
     status = Column(String(256), nullable=False)