chia-blockchain/chia/cmds/data.py

from __future__ import annotations

import json
import logging
from collections.abc import Callable, Coroutine, Sequence
from pathlib import Path
from typing import Any, TypeVar

import click
from chia_rs.sized_bytes import bytes32
from chia_rs.sized_ints import uint64

from chia.cmds import options
from chia.cmds.param_types import Bytes32ParamType

FC = TypeVar("FC", bound=Callable[..., Any] | click.Command)

logger = logging.getLogger(__name__)


# TODO: this is more general and should be part of refactoring the overall CLI code duplication
def run(coro: Coroutine[Any, Any, dict[str, Any] | None]) -> None:
    import asyncio

    response = asyncio.run(coro)

    success = response is not None and response.get("success", False)
    logger.info(f"data layer cli call response:{success}")
    # todo make sure all cli methods follow this pattern, uncomment
    # if not success:
    # raise click.ClickException(message=f"query unsuccessful, response: {response}")


@click.group("data", help="Manage your data")
def data_cmd() -> None:
    pass


# TODO: maybe use more helpful `type=`s to get click to handle error reporting of
#       malformed inputs.


def create_changelist_option() -> Callable[[FC], FC]:
    return click.option(
        "-d",
        "--changelist",
        "changelist_string",
        help="str representing the changelist",
        type=str,
        required=True,
    )


def create_store_updates_option() -> Callable[[FC], FC]:
    return click.option(
        "-d",
        "--store_updates",
        "store_updates_string",
        help="str representing the store updates",
        type=str,
        required=True,
    )


def create_key_option(multiple: bool = False) -> Callable[[FC], FC]:
    return click.option(
        "-k",
        "--key",
        "key_strings" if multiple else "key_string",
        help="str representing the key",
        type=str,
        required=True,
        multiple=multiple,
    )


def create_data_store_id_option() -> Callable[[FC], FC]:
    return click.option(
        "-store",
        "--id",
        help="The hexadecimal store id.",
        type=Bytes32ParamType(),
        required=True,
    )


def create_data_store_name_option() -> Callable[[FC], FC]:
    return click.option(
        "-n",
        "--table_name",
        "table_name",
        help="The name of the table.",
        type=str,
        required=True,
    )


def create_rpc_port_option() -> Callable[[FC], FC]:
    return click.option(
        "-dp",
        "--data-rpc-port",
        help="Set the port where the data layer is hosting the RPC interface. See rpc_port under wallet in config.yaml",
        type=int,
        default=None,
        show_default=True,
    )


def create_root_hash_option() -> Callable[[FC], FC]:
    return click.option(
        "-r",
        "--root_hash",
        help="The hexadecimal root hash",
        type=Bytes32ParamType(),
        required=False,
    )


def create_page_option() -> Callable[[FC], FC]:
    return click.option(
        "-p",
        "--page",
        help="Enables pagination of the output and requests a specific page.",
        type=int,
        required=False,
    )


def create_max_page_size_option() -> Callable[[FC], FC]:
    return click.option(
        "--max-page-size",
        help="Set how many bytes to be included in a page, if pagination is enabled.",
        type=int,
        required=False,
    )


# Functions with this mark in this file are not being ported to @tx_out_cmd due to API peculiarities
# They will therefore not work with observer-only functionality
# NOTE: tx_endpoint  (This creates wallet transactions and should be parametrized by relevant options)
@data_cmd.command("create_data_store", help="Create a new data store")
@create_rpc_port_option()
@options.create_fee()
@click.option("--verbose", is_flag=True, help="Enable verbose output.")
@options.create_fingerprint()
def create_data_store(
    data_rpc_port: int,
    fee: uint64 | None,
    verbose: bool,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import create_data_store_cmd

    run(create_data_store_cmd(data_rpc_port, fee, verbose, fingerprint=fingerprint))


@data_cmd.command("get_value", help="Get the value for a given key and store")
@create_data_store_id_option()
@create_key_option()
@create_root_hash_option()
@create_rpc_port_option()
@options.create_fingerprint()
def get_value(
    id: bytes32,
    key_string: str,
    root_hash: bytes32 | None,
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import get_value_cmd

    run(get_value_cmd(data_rpc_port, id, key_string, root_hash, fingerprint=fingerprint))


# NOTE: tx_endpoint
@data_cmd.command("update_data_store", help="Update a store by providing the changelist operations")
@create_data_store_id_option()
@create_changelist_option()
@create_rpc_port_option()
@options.create_fee()
@options.create_fingerprint()
@click.option("--submit/--no-submit", default=True, help="Submit the result on chain")
def update_data_store(
    id: bytes32,
    changelist_string: str,
    data_rpc_port: int,
    fee: uint64 | None,
    fingerprint: int | None,
    submit: bool,
) -> None:
    from chia.cmds.data_funcs import update_data_store_cmd

    run(
        update_data_store_cmd(
            rpc_port=data_rpc_port,
            store_id=id,
            changelist=json.loads(changelist_string),
            fee=fee,
            fingerprint=fingerprint,
            submit_on_chain=submit,
        )
    )


@data_cmd.command("update_multiple_stores", help="Update multiple stores by providing the changelist operations")
@create_store_updates_option()
@create_rpc_port_option()
@options.create_fee()
@options.create_fingerprint()
@click.option("--submit/--no-submit", default=True, help="Submit the result on chain")
def update_multiple_stores(
    store_updates_string: str,
    data_rpc_port: int,
    fee: uint64,
    fingerprint: int | None,
    submit: bool,
) -> None:
    from chia.cmds.data_funcs import update_multiple_stores_cmd

    run(
        update_multiple_stores_cmd(
            rpc_port=data_rpc_port,
            store_updates=json.loads(store_updates_string),
            fee=fee,
            fingerprint=fingerprint,
            submit_on_chain=submit,
        )
    )


@data_cmd.command("submit_pending_root", help="Submit on chain a locally stored batch")
@create_data_store_id_option()
@create_rpc_port_option()
@options.create_fee()
@options.create_fingerprint()
def submit_pending_root(
    id: bytes32,
    data_rpc_port: int,
    fee: uint64,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import submit_pending_root_cmd

    run(
        submit_pending_root_cmd(
            rpc_port=data_rpc_port,
            store_id=id,
            fee=fee,
            fingerprint=fingerprint,
        )
    )


@data_cmd.command("submit_all_pending_roots", help="Submit on chain all locally stored batches")
@create_rpc_port_option()
@options.create_fee()
@options.create_fingerprint()
def submit_all_pending_roots(
    data_rpc_port: int,
    fee: uint64,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import submit_all_pending_roots_cmd

    run(
        submit_all_pending_roots_cmd(
            rpc_port=data_rpc_port,
            fee=fee,
            fingerprint=fingerprint,
        )
    )


@data_cmd.command("get_keys", help="Get all keys for a given store")
@create_data_store_id_option()
@create_root_hash_option()
@create_rpc_port_option()
@options.create_fingerprint()
@create_page_option()
@create_max_page_size_option()
def get_keys(
    id: bytes32,
    root_hash: bytes32 | None,
    data_rpc_port: int,
    fingerprint: int | None,
    page: int | None,
    max_page_size: int | None,
) -> None:
    from chia.cmds.data_funcs import get_keys_cmd

    run(get_keys_cmd(data_rpc_port, id, root_hash, fingerprint=fingerprint, page=page, max_page_size=max_page_size))


@data_cmd.command("get_keys_values", help="Get all keys and values for a given store")
@create_data_store_id_option()
@create_root_hash_option()
@create_rpc_port_option()
@options.create_fingerprint()
@create_page_option()
@create_max_page_size_option()
def get_keys_values(
    id: bytes32,
    root_hash: bytes32 | None,
    data_rpc_port: int,
    fingerprint: int | None,
    page: int | None,
    max_page_size: int | None,
) -> None:
    from chia.cmds.data_funcs import get_keys_values_cmd

    run(
        get_keys_values_cmd(
            data_rpc_port, id, root_hash, fingerprint=fingerprint, page=page, max_page_size=max_page_size
        )
    )


@data_cmd.command("get_root", help="Get the published root hash value for a given store")
@create_data_store_id_option()
@create_rpc_port_option()
@options.create_fingerprint()
def get_root(
    id: bytes32,
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import get_root_cmd

    run(get_root_cmd(rpc_port=data_rpc_port, store_id=id, fingerprint=fingerprint))


@data_cmd.command("subscribe", help="Subscribe to a store")
@create_data_store_id_option()
@click.option(
    "-u",
    "--url",
    "urls",
    help="Manually provide a list of servers urls for downloading the data",
    type=str,
    multiple=True,
)
@create_rpc_port_option()
@options.create_fingerprint()
def subscribe(
    id: bytes32,
    urls: list[str],
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import subscribe_cmd

    run(subscribe_cmd(rpc_port=data_rpc_port, store_id=id, urls=urls, fingerprint=fingerprint))


@data_cmd.command("remove_subscription", help="Remove server urls that are added via subscribing to urls")
@create_data_store_id_option()
@click.option("-u", "--url", "urls", help="Server urls to remove", type=str, multiple=True)
@create_rpc_port_option()
@options.create_fingerprint()
def remove_subscription(
    id: bytes32,
    urls: list[str],
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import remove_subscriptions_cmd

    run(remove_subscriptions_cmd(rpc_port=data_rpc_port, store_id=id, urls=urls, fingerprint=fingerprint))


@data_cmd.command("unsubscribe", help="Completely untrack a store")
@create_data_store_id_option()
@create_rpc_port_option()
@options.create_fingerprint()
@click.option("--retain", is_flag=True, help="Retain .dat files")
def unsubscribe(
    id: bytes32,
    data_rpc_port: int,
    fingerprint: int | None,
    retain: bool,
) -> None:
    from chia.cmds.data_funcs import unsubscribe_cmd

    run(unsubscribe_cmd(rpc_port=data_rpc_port, store_id=id, fingerprint=fingerprint, retain=retain))


@data_cmd.command(
    "get_kv_diff", help="Get the inserted and deleted keys and values between an initial and a final hash"
)
@create_data_store_id_option()
@click.option("-hash_1", "--hash_1", help="Initial hash", type=Bytes32ParamType(), required=True)
@click.option("-hash_2", "--hash_2", help="Final hash", type=Bytes32ParamType(), required=True)
@create_rpc_port_option()
@options.create_fingerprint()
@create_page_option()
@create_max_page_size_option()
def get_kv_diff(
    id: bytes32,
    hash_1: bytes32,
    hash_2: bytes32,
    data_rpc_port: int,
    fingerprint: int | None,
    page: int | None,
    max_page_size: int | None,
) -> None:
    from chia.cmds.data_funcs import get_kv_diff_cmd

    run(
        get_kv_diff_cmd(
            rpc_port=data_rpc_port,
            store_id=id,
            hash_1=hash_1,
            hash_2=hash_2,
            fingerprint=fingerprint,
            page=page,
            max_page_size=max_page_size,
        )
    )


@data_cmd.command("get_root_history", help="Get all changes of a singleton")
@create_data_store_id_option()
@create_rpc_port_option()
@options.create_fingerprint()
def get_root_history(
    id: bytes32,
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import get_root_history_cmd

    run(get_root_history_cmd(rpc_port=data_rpc_port, store_id=id, fingerprint=fingerprint))


@data_cmd.command("add_missing_files", help="Manually reconstruct server files from the data layer database")
@click.option(
    "-i",
    "--ids",
    help="List of stores to reconstruct. If not specified, all stores will be reconstructed",
    type=str,
    multiple=True,
    required=False,
)
@click.option(
    "-o/-n",
    "--overwrite/--no-overwrite",
    help="Specify if already existing files need to be overwritten by this command",
)
@click.option(
    "-d", "--directory", type=str, help="If specified, use a non-default directory to write the files", required=False
)
@create_rpc_port_option()
@options.create_fingerprint()
def add_missing_files(
    ids: Sequence[bytes32],
    overwrite: bool,
    directory: str | None,
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import add_missing_files_cmd

    run(
        add_missing_files_cmd(
            rpc_port=data_rpc_port,
            ids=list(ids) if ids else None,
            overwrite=overwrite,
            foldername=None if directory is None else Path(directory),
            fingerprint=fingerprint,
        )
    )


# NOTE: tx_endpoint
@data_cmd.command("add_mirror", help="Publish mirror urls on chain")
@create_data_store_id_option()
@click.option(
    "-a", "--amount", help="Amount to spend for this mirror, in mojos", type=int, default=0, show_default=True
)
@click.option(
    "-u",
    "--url",
    "urls",
    help="URL to publish on the new coin, multiple accepted and will be published to a single coin.",
    type=str,
    multiple=True,
)
@options.create_fee()
@create_rpc_port_option()
@options.create_fingerprint()
def add_mirror(
    id: bytes32,
    amount: int,
    urls: list[str],
    fee: uint64 | None,
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import add_mirror_cmd

    run(
        add_mirror_cmd(
            rpc_port=data_rpc_port,
            store_id=id,
            urls=urls,
            amount=amount,
            fee=fee,
            fingerprint=fingerprint,
        )
    )


# NOTE: tx_endpoint
@data_cmd.command("delete_mirror", help="Delete an owned mirror by its coin id")
@click.option("-c", "--coin_id", help="Coin id", type=Bytes32ParamType(), required=True)
@options.create_fee()
@create_rpc_port_option()
@options.create_fingerprint()
def delete_mirror(
    coin_id: bytes32,
    fee: uint64 | None,
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import delete_mirror_cmd

    run(
        delete_mirror_cmd(
            rpc_port=data_rpc_port,
            coin_id=coin_id,
            fee=fee,
            fingerprint=fingerprint,
        )
    )


@data_cmd.command("get_mirrors", help="Get a list of all mirrors for a given store")
@create_data_store_id_option()
@create_rpc_port_option()
@options.create_fingerprint()
def get_mirrors(
    id: bytes32,
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import get_mirrors_cmd

    run(
        get_mirrors_cmd(
            rpc_port=data_rpc_port,
            store_id=id,
            fingerprint=fingerprint,
        )
    )


@data_cmd.command("get_subscriptions", help="Get subscribed stores, including the owned stores")
@create_rpc_port_option()
@options.create_fingerprint()
def get_subscriptions(
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import get_subscriptions_cmd

    run(
        get_subscriptions_cmd(
            rpc_port=data_rpc_port,
            fingerprint=fingerprint,
        )
    )


@data_cmd.command("get_owned_stores", help="Get owned stores")
@create_rpc_port_option()
@options.create_fingerprint()
def get_owned_stores(
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import get_owned_stores_cmd

    run(
        get_owned_stores_cmd(
            rpc_port=data_rpc_port,
            fingerprint=fingerprint,
        )
    )


@data_cmd.command("get_sync_status", help="Get locally stored root compared to the root of the singleton")
@create_data_store_id_option()
@create_rpc_port_option()
@options.create_fingerprint()
def get_sync_status(
    id: bytes32,
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import get_sync_status_cmd

    run(get_sync_status_cmd(rpc_port=data_rpc_port, store_id=id, fingerprint=fingerprint))


@data_cmd.group("plugins", help="Get information about configured uploader/downloader plugins")
def plugins_cmd() -> None:
    pass


@plugins_cmd.command("check", help="Calls the plugin_info endpoint on all configured plugins")
@create_rpc_port_option()
def check_plugins(
    data_rpc_port: int,
) -> None:
    from chia.cmds.data_funcs import check_plugins_cmd

    run(check_plugins_cmd(rpc_port=data_rpc_port))


@data_cmd.command(
    "clear_pending_roots",
    help="Clear pending roots that will not be published, associated data may not be recoverable",
)
@create_data_store_id_option()
@click.confirmation_option(
    prompt="Associated data may not be recoverable.\nAre you sure you want to remove the pending roots?",
)
@create_rpc_port_option()
@options.create_fingerprint()
def clear_pending_roots(
    id: bytes32,
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import clear_pending_roots

    run(
        clear_pending_roots(
            rpc_port=data_rpc_port,
            store_id=id,
            fingerprint=fingerprint,
        )
    )


@data_cmd.command(
    "wallet_log_in",
    help="Request that the wallet service be logged in to the specified fingerprint",
)
@create_rpc_port_option()
@options.create_fingerprint(required=True)
def wallet_log_in(
    data_rpc_port: int,
    fingerprint: int,
) -> None:
    from chia.cmds.data_funcs import wallet_log_in_cmd

    run(
        wallet_log_in_cmd(
            rpc_port=data_rpc_port,
            fingerprint=fingerprint,
        )
    )


@data_cmd.command(
    "get_proof",
    help="Obtains a merkle proof of inclusion for a given key",
)
@create_data_store_id_option()
@create_rpc_port_option()
@create_key_option(multiple=True)
@options.create_fingerprint()
def get_proof(
    id: bytes32,
    key_strings: list[str],
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import get_proof_cmd

    run(get_proof_cmd(rpc_port=data_rpc_port, store_id=id, fingerprint=fingerprint, key_strings=key_strings))


@data_cmd.command(
    "verify_proof",
    help="Verifies a merkle proof of inclusion",
)
@click.option(
    "-p",
    "--proof",
    "proof_string",
    help="Proof to validate in JSON format.",
    type=str,
)
@create_rpc_port_option()
@options.create_fingerprint()
def verify_proof(
    proof_string: str,
    data_rpc_port: int,
    fingerprint: int | None,
) -> None:
    from chia.cmds.data_funcs import verify_proof_cmd

    proof_dict = json.loads(proof_string)
    run(verify_proof_cmd(rpc_port=data_rpc_port, fingerprint=fingerprint, proof=proof_dict))