blockstream-satellite-api/server/transmitter.py

import json
import logging
from datetime import datetime

from flask import current_app
from sqlalchemy import and_

import constants
import order_helpers
from database import db
from models import Order, TxRetry
from regions import region_code_to_number_list
from schemas import order_schema


def assign_tx_seq_num(order):
    """Assign Tx sequence number to order"""
    last_tx_order = Order.query.order_by(Order.tx_seq_num.desc()).first()

    if last_tx_order.tx_seq_num:
        order.tx_seq_num = last_tx_order.tx_seq_num + 1
    else:
        order.tx_seq_num = 1
    db.session.commit()


def redis():
    return current_app.config.get("REDIS_INSTANCE")


def publish_to_sse_server(order, retransmit_info=None):
    msg = order_schema.dump(order)
    # If it's a retransmission, take the regions list from the tx_retries table
    # instead of the orders table.
    if retransmit_info:
        msg['regions'] = region_code_to_number_list(
            retransmit_info.region_code)
    msg = json.dumps(msg)
    redis().publish(channel=constants.CHANNEL_INFO[order.channel].name,
                    message=msg)
    return


def tx_start(channel=None):
    """Look for pending transmissions and serve them

    An order is ready for transmission when already paid or when being
    retransmitted. Also, a pending transmission can only be served if there is
    no other ongoing transmission on the logical channel. Each channel can only
    serve one transmission at a time.

    This function works both for a single defined channel or for all channels.
    When the channel parameter is undefined, it looks for pending transmissions
    in all channels. Otherwise, it processes the specified channel only.

    Args:
        channel (int, optional): Logical transmission channel to serve.
            Defaults to None.

    """
    # Call itself recursively if the channel is not defined
    if (channel is None):
        for channel in constants.CHANNELS:
            tx_start(channel)
        return

    transmitting_orders = Order.query.filter(
        and_(Order.status == constants.OrderStatus.transmitting.value,
             Order.channel == channel)).all()

    # Do not start a new transmission if another order is being transmitted
    # right now
    if len(transmitting_orders) > 0:
        return

    # First, try to find a paid order with the highest bid in the orders table
    # and start its transmission. If no paid orders are found there, look into
    # the tx_retries table and retransmit one of the orders from there if it
    # meets the retransmission criteria
    order = Order.query.filter(
        and_(Order.status == constants.OrderStatus.paid.value,
             Order.channel == channel)).order_by(
                 Order.bid_per_byte.desc()).first()

    if order:
        logging.info(f'transmission start {order.uuid}')
        assign_tx_seq_num(order)
        order.status = constants.OrderStatus.transmitting.value
        order.started_transmission_at = datetime.utcnow()
        db.session.commit()
        publish_to_sse_server(order)
    else:
        # No order found for the first transmission.
        # Check if any order requires retransmission.
        order, retransmit_info = order_helpers.get_next_retransmission(channel)
        if order and retransmit_info:
            logging.info(f'retransmission start {order.uuid}')
            order.status = constants.OrderStatus.transmitting.value
            retransmit_info.retry_count += 1
            retransmit_info.last_attempt = datetime.utcnow()
            retransmit_info.pending = False
            db.session.commit()
            publish_to_sse_server(order, retransmit_info)


def tx_end(order):
    """End transmission"""
    if order.ended_transmission_at is None:
        logging.info(f'transmission end {order.uuid}')
        order.ended_transmission_at = datetime.utcnow()
        retransmit_info = TxRetry.query.filter_by(order_id=order.id).first()
        # Cleanup the TxRetry
        TxRetry.query.filter_by(order_id=order.id).delete()
        db.session.commit()
        publish_to_sse_server(order, retransmit_info)
        # Start the next queued order as soon as the current order finishes
        tx_start(order.channel)