use std::collections::HashMap;

use std::future::Future;

use std::sync::Arc;

use actix::fut::future::LocalBoxActorFuture;

use actix::prelude::*;

use tokio::sync::RwLock;

use tracing::instrument;

use crate::image_resolver::ImageResolver;

use crate::utils::{LVMStoreFactory, bartender::DeployInfo};

use lyquor_api::{

    actor::{FetchOracleInfo, GetAddressByEd25519, GetEd25519ByAddress, GetEthContractAddr, Stop, recipient_service},

    call::CallParams,

};

use lyquor_primitives::{Address, ChainPos, LyquidID, NodeID, SequenceBackendID, encode_by_fields};

use lyquor_seq::{StopHostedLyquid, fco::FCO, side_effects::SideEffectStore};

use lyquor_vm::scheduler::{RunOptions, RunSource};

use super::{

    Error, GetBartender, GetConsoleBuffer, GetConsoleBufferResult, GetDeployedLyquidList, GetHostedLyquidList,

    GetIdByContract, GetLatestLyquidDeploymentInfo, GetLyquid, GetLyquidByEthAddr, GetLyquidDeploymentInfo,

    GetRegisteredLyquidList, ImageRepo, ImageRepoResponse, LVM, LoadLyquid, Lyquid, LyquidConfig, LyquidNumber,

    LyquidProcess, SetPoolAddr, StopHostedLyquids, UnloadLyquid,

};

impl Handler<GetAddressByEd25519> for LyquidPool {

    type Result = ResponseFuture<Option<Address>>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: GetAddressByEd25519, _ctx: &mut Context<Self>) -> Self::Result {

        Box::pin(self.get_address_by_ed25519(msg.pos, msg.id))

    }

}

impl Handler<GetEd25519ByAddress> for LyquidPool {

    type Result = ResponseFuture<Option<NodeID>>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: GetEd25519ByAddress, _ctx: &mut Context<Self>) -> Self::Result {

        Box::pin(self.get_ed25519_by_address(msg.pos, msg.address))

    }

}

impl Handler<GetEthContractAddr> for LyquidPool {

    type Result = ResponseFuture<Option<Address>>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: lyquor_api::actor::GetEthContractAddr, _ctx: &mut Context<Self>) -> Self::Result {

        Box::pin(self.get_eth_contract_address(msg.pos, msg.id))

    }

}

impl Handler<FetchOracleInfo> for LyquidPool {

    type Result = ResponseFuture<Option<lyquor_primitives::oracle::OracleEpochInfo>>;

    #[tracing::instrument(level = "trace", skip(self, msg, _ctx))]

    fn handle(&mut self, msg: FetchOracleInfo, _ctx: &mut Context<Self>) -> Self::Result {

        let id = match msg.target.target {

            lyquor_primitives::oracle::OracleServiceTarget::LVM(id) => id,

            lyquor_primitives::oracle::OracleServiceTarget::EVM { .. } => return Box::pin(std::future::ready(None)),

        };

        let lyquid = self.lyquid(&id);

        Box::pin(async move {

            let lyquid = lyquid?;

            let ln = lyquid.get_lyquid_number(msg.pos).await.ok().flatten()?;

            let call = lyquid

                .instance

                .call_instance_func_decoded(

                    ln,

                    CallParams::builder()

                        .caller(Address::ZERO)

                        .method("__lyquor_oracle_dest_epoch_info".into())

                        .input(encode_by_fields!(topic: String = msg.topic, full_config: bool = msg.full_config).into())

                        .build(),

                    RunOptions::new(RunSource::InstanceCall),

                )

                .await;

            let (epoch, config_hash, change_count, config_raw): (

                u64,

                lyquor_primitives::B256,

                u32,

                lyquor_primitives::Bytes,

            ) = call.await.ok()?;

            let epoch = u32::try_from(epoch).ok()?;

            let config = if config_raw.is_empty() {

                None

            } else {

                lyquor_primitives::decode_object(config_raw.as_ref())

            };

            Some(lyquor_primitives::oracle::OracleEpochInfo {

                epoch,

                config_hash: <[u8; 32]>::from(config_hash).into(),

                change_count,

                config,

            })

        })

    }

}

impl Handler<GetBartender> for LyquidPool {

    type Result = Lyquid;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, _: GetBartender, _ctx: &mut Context<Self>) -> Self::Result {

        self.bartender.clone()

    }

}

impl Handler<GetLyquid> for LyquidPool {

    type Result = Option<Lyquid>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: GetLyquid, _ctx: &mut Context<Self>) -> Self::Result {

        self.lyquid(&msg.id)

    }

}

impl Handler<GetIdByContract> for LyquidPool {

    type Result = ResponseFuture<Option<LyquidID>>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: GetIdByContract, _ctx: &mut Context<Self>) -> Self::Result {

        Box::pin(self.id_by_contract(msg.addr))

    }

}

impl Handler<GetLyquidByEthAddr> for LyquidPool {

    type Result = AtomicResponse<Self, Option<(Lyquid, LyquidNumber)>>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: GetLyquidByEthAddr, _ctx: &mut Context<Self>) -> Self::Result {

        AtomicResponse::new(self.lyquid_by_eth_addr(msg.addr, msg.pos))

    }

}

impl Handler<GetRegisteredLyquidList> for LyquidPool {

    type Result = ResponseFuture<Option<Vec<(LyquidID, Vec<LyquidID>)>>>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, _: GetRegisteredLyquidList, _ctx: &mut Context<Self>) -> Self::Result {

        Box::pin(self.registered_lyquid_list())

    }

}

impl Handler<GetLyquidDeploymentInfo> for LyquidPool {

    type Result = ResponseFuture<Option<DeployInfo>>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: GetLyquidDeploymentInfo, _ctx: &mut Context<Self>) -> Self::Result {

        Box::pin(self.lyquid_deployment_info(msg.id, msg.idx))

    }

}

impl Handler<GetLatestLyquidDeploymentInfo> for LyquidPool {

    type Result = ResponseFuture<Option<DeployInfo>>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: GetLatestLyquidDeploymentInfo, _ctx: &mut Context<Self>) -> Self::Result {

        Box::pin(self.latest_lyquid_deployment_info(msg.id))

    }

}

impl Handler<GetHostedLyquidList> for LyquidPool {

    type Result = Vec<LyquidID>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, _: GetHostedLyquidList, _ctx: &mut Context<Self>) -> Self::Result {

        self.pool.keys().cloned().collect()

    }

}

impl Handler<GetDeployedLyquidList> for LyquidPool {

    type Result = ResponseFuture<Vec<LyquidID>>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, _: GetDeployedLyquidList, _ctx: &mut Context<Self>) -> Self::Result {

        Box::pin(self.deployed_lyquid_list())

    }

}

impl Handler<ImageRepo> for LyquidPool {

    type Result = ImageRepoResponse;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, _: ImageRepo, _ctx: &mut Context<Self>) -> Self::Result {

        ImageRepoResponse(self.setup.image_resolver.image_repo())

    }

}

impl Handler<GetConsoleBuffer> for LyquidPool {

    type Result = ResponseFuture<Option<GetConsoleBufferResult>>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: GetConsoleBuffer, _ctx: &mut Context<Self>) -> Self::Result {

        match self.pool.get(&msg.id) {

            Some(lyquid) => {

                let proc = lyquid.proc.clone();

                Box::pin(async move { proc.send(msg).await.ok().flatten() })

            }

            None => Box::pin(std::future::ready(None)),

        }

    }

}

impl Handler<LoadLyquid> for LyquidPool {

    type Result = AtomicResponse<Self, Result<bool, Error>>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: LoadLyquid, ctx: &mut Context<Self>) -> Self::Result {

        let LoadLyquid { id, deps } = msg;

        if self.pool.contains_key(&id) {

            // TODO: This does not handle live Lyquid upgrades. If an already-hosted Lyquid is

            // redeployed with a new image digest, we return early here and never prefetch that

            // image before the upgraded slot arrives.

            return AtomicResponse::new(Box::pin(async move { Ok(true) }.into_actor(self)));

        }

        let deploy_info = self.latest_lyquid_deployment_info(id);

        let image_resolver = self.setup.image_resolver.clone();

        let lvm = self.lvm.clone();

        let setup = self.setup.clone();

        let bartender = Some(lyquor_vm::Bartender {

            get_address_by_ed25519: recipient_service(ctx.address().recipient()),

            get_ed25519_by_address: recipient_service(ctx.address().recipient()),

            get_eth_contract_address: recipient_service(ctx.address().recipient()),

            sequence_backend_id: self.sequence_backend_id,

        });

        let oracle_info = Some(recipient_service(ctx.address().recipient()));

        AtomicResponse::new(Box::pin(

            async move {

                // Non-bartender lyquids are registered before their first constructor slot,

                // so bartender metadata is available in time to prefetch the latest image here.

                if let Some(deploy_info) = deploy_info.await {

                    if let Some(repo_hint) = deploy_info.repo_hint.as_deref() {

                        image_resolver.remember_repository_hint(deploy_info.image_digest, repo_hint);

                    }

                    if let Err(e) = image_resolver.ensure_local(deploy_info.image_digest).await {

                        tracing::debug!("Skipping lyquid image prefetch after resolution failure: {e}");

                    }

                }

                LyquidProcess::start(id, deps, lvm, setup, bartender, oracle_info).await

            }

            .into_actor(self)

            .map(move |result, act, ctx| {

                let entry = result?;

                entry.proc.do_send(SetPoolAddr { pool: ctx.address() });

                act.pool.insert(id, entry);

                Ok(false)

            }),

        ))

    }

}

impl Handler<UnloadLyquid> for LyquidPool {

    type Result = Result<(), Error>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: UnloadLyquid, _ctx: &mut Context<Self>) -> Self::Result {

        if msg.id == *self.bartender.id() {

            return Err(Error::CannotUnloadBartender);

        }

        let Some(lyquid) = self.pool.remove(&msg.id) else {

            return Err(Error::LyquidNotHosted);

        };

        lyquid.proc.do_send(Stop);

        Ok(())

    }

}

impl Handler<StopHostedLyquids> for LyquidPool {

    type Result = ();

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, msg: StopHostedLyquids, _ctx: &mut Context<Self>) -> Self::Result {

        for id in msg.ids {

            if let Some(lyquid) = self.pool.get(&id) {

                lyquid.proc.do_send(StopHostedLyquid);

            }

        }

    }

}

impl Handler<Stop> for LyquidPool {

    type Result = AtomicResponse<Self, ()>;

    #[tracing::instrument(level = "trace")]

    fn handle(&mut self, _: Stop, _ctx: &mut Context<Self>) -> Self::Result {

        let procs = self

            .pool

            .iter()

            .map(|(_, lyquid)| lyquid.proc.clone())

            .collect::<Vec<_>>();

        AtomicResponse::new(Box::pin(

            async move {

                for proc in procs {

                    proc.send(Stop).await.ok();

                }

            }

            .into_actor(self)

            .map(|_, _act, ctx| {

                ctx.stop();

            }),

        ))

    }

}

#[derive(Clone)]

pub struct LyquidPoolSetup {

    pub fco: Addr<FCO>,

    pub store_factory: Arc<dyn LVMStoreFactory>,

    pub image_resolver: Arc<ImageResolver>,

    pub side_effect_store: Arc<SideEffectStore>,

    pub log_hub: crate::lyquid_log::LyquidLogHandle,

    pub sequence_backend_chain_id: u64,

    pub sequence_backend_bartender_addr: Address,

    pub config: LyquidConfig,

}

/// Collection of all serving Lyquids.

pub struct LyquidPool {

    pool: HashMap<LyquidID, Lyquid>,

    lvm: Arc<RwLock<LVM>>,

    bartender: Lyquid,

    sequence_backend_id: SequenceBackendID,

    setup: LyquidPoolSetup,

}

lyquor_primitives::debug_struct_name!(LyquidPool);

impl Actor for LyquidPool {

    type Context = Context<Self>;

    #[instrument(level = "trace", skip(ctx))]

    fn started(&mut self, ctx: &mut Self::Context) {

        let pool = ctx.address();

        for lyquid in self.pool.values() {

            lyquid.proc.do_send(SetPoolAddr { pool: pool.clone() });

        }

    }

    #[instrument(level = "trace", skip(_ctx))]

    fn stopped(&mut self, _ctx: &mut Self::Context) {}

}

impl LyquidPool {

    pub async fn new(bartender_id: LyquidID, lvm: LVM, setup: LyquidPoolSetup) -> Result<Self, Error> {

        let lvm = Arc::new(RwLock::new(lvm));

        let bartender = LyquidProcess::start(bartender_id, vec![], lvm.clone(), setup.clone(), None, None).await
?0
;

        let bartender_handle = bartender.clone();

        let sequence_backend_id = lyquor_primitives::sequence_backend_id(

            setup.sequence_backend_chain_id,

            setup.sequence_backend_bartender_addr,

        );

        let mut pool = HashMap::new();

        pool.insert(bartender_id, bartender);

        Ok(Self {

            bartender: bartender_handle,

            pool,

            lvm,

            sequence_backend_id,

            setup,

        })

    }

    fn lyquid(&self, id: &LyquidID) -> Option<Lyquid> {

        self.pool.get(id).map(|l| l.clone())

    }

    fn id_by_contract(&self, addr: Address) -> impl Future<Output = Option<LyquidID>> + 'static {

        let bartender = self.bartender.instance().clone();

        async move {

            // TODO: proper caching to avoid calling bartender every time

            let call = {

                // We only need to write-lock the instance shortly to generate the async call task. Do

                // not make await the call right away because we don't need to hold the Instance lock

                // to execute the call.

                let bar_ln = bartender.latest_number().await;

                bartender

                    .call_instance_func_decoded(

                        bar_ln,

                        CallParams::builder()

                            .caller(Address::ZERO)

                            .method("get_lyquid_id_by_eth_addr".into())

                            .input(encode_by_fields!(addr: Address = addr).into())

                            .build(),

                        RunOptions::new(RunSource::Bartender),

                    )

                    .await

            };

            let id: Option<LyquidID> = call.await.ok()?;

            id

        }

    }

    fn lyquid_by_eth_addr(

        &mut self, addr: Address, pos: ChainPos,

    ) -> LocalBoxActorFuture<Self, Option<(Lyquid, LyquidNumber)>> {

        let bartender = self.bartender.clone();

        // TODO: proper caching to avoid calling bartender every time

        Box::pin(self.id_by_contract(addr).into_actor(self).then(move |id, act, _ctx| {

            let lyquid = id.and_then(|id| act.lyquid(&id));

            async move {

                let lyquid = lyquid?;

                // Now we should check if such contract address is valid given the chain position. Because

                // the contract could be superseded by a later contract for the same lyquid, if the

                // contract address at the given chain position does not match the address in effect, we

                // should not regard this as a valid action.

                let ln = lyquid.get_lyquid_number(pos).await.ok()??;

                let contract = Self::bartender_get_eth_addr(bartender, pos, lyquid.id, ln.image).await;

                if contract? == addr { Some((lyquid, ln)) } else { None }

            }

            .into_actor(act)

        }))

    }

    fn registered_lyquid_list(&self) -> impl Future<Output = Option<Vec<(LyquidID, Vec<LyquidID>)>>> + 'static {

        let bartender = self.bartender.instance().clone();

        async move {

            let call = {

                // We only need to write-lock the instance shortly to generate the async call task. Do

                // not make await the call right away because we don't need to hold the Instance lock

                // to execute the call.

                let bar_ln = bartender.latest_number().await;

                // TODO: get_lyquid_list_with_deps() is not scalable even if the node wants to subscribe to all

                // lyquids. We need to change it to an iterator style invcation.

                bartender

                    .call_instance_func_decoded(

                        bar_ln,

                        CallParams::builder()

                            .caller(Address::ZERO)

                            .method("get_lyquid_list_with_deps".into())

                            .input(encode_by_fields!().into())

                            .build(),

                        RunOptions::new(RunSource::Bartender),

                    )

                    .await

            };

            let result: Vec<(LyquidID, Vec<LyquidID>)> = call.await.ok()?;

            // NOTE: we need to explicitly declare result for correct decoding

            Some(result)

        }

    }

    fn lyquid_deployment_info(&self, id: LyquidID, nth: u32) -> impl Future<Output = Option<DeployInfo>> + 'static {

        let bartender = self.bartender.instance().clone();

        async move {

            let call = {

                // We only need to write-lock the instance shortly to generate the async call task. Do

                // not make await the call right away because we don't need to hold the Instance lock

                // to execute the call.

                let bar_ln = bartender.latest_number().await;

                // TODO: get_lyquid_list() is not scalable even if the node wants to subscribe to all

                // lyquids. We need to change it to an iterator style invcation.

                bartender

                    .call_instance_func_decoded(

                        bar_ln,

                        CallParams::builder()

                            .caller(Address::ZERO)

                            .method("get_lyquid_deployment_info".into())

                            .input(encode_by_fields!(id: LyquidID = id, nth: u32 = nth).into())

                            .build(),

                        RunOptions::new(RunSource::Bartender),

                    )

                    .await

            };

            let result: Option<DeployInfo> = call.await.ok()?;

            // NOTE: we need to explicitly declare result for correct decoding

            result

        }

    }

    fn latest_lyquid_deployment_info(&self, id: LyquidID) -> impl Future<Output = Option<DeployInfo>> + 'static {

        let bartender = self.bartender.instance().clone();

        async move {

            let call = {

                let bar_ln = bartender.latest_number().await;

                bartender

                    .call_instance_func_decoded(

                        bar_ln,

                        CallParams::builder()

                            .caller(Address::ZERO)

                            .method("get_last_lyquid_deployment_info".into())

                            .input(encode_by_fields!(id: LyquidID = id).into())

                            .build(),

                        RunOptions::new(RunSource::Bartender),

                    )

                    .await

            };

            let result: Option<DeployInfo> = call.await.ok()?;

            result

        }

    }

    fn deployed_lyquid_list(&self) -> impl Future<Output = Vec<LyquidID>> + 'static {

        let bartender = self.bartender.instance().clone();

        async move {

            let call = {

                // We only need to write-lock the instance shortly to generate the async call task. Do

                // not make await the call right away because we don't need to hold the Instance lock

                // to execute the call.

                let bar_ln = bartender.latest_number().await;

                bartender

                    .call_instance_func_decoded(

                        bar_ln,

                        CallParams::builder()

                            .caller(Address::ZERO)

                            .method("get_lyquid_list".into())

                            .input(encode_by_fields!().into())

                            .build(),

                        RunOptions::new(RunSource::Bartender),

                    )

                    .await

            };

            let resp: Option<Vec<LyquidID>> = call.await.ok();

            resp.unwrap_or_default()

        }

    }

    fn get_eth_contract_address(&self, pos: ChainPos, id: LyquidID) -> impl Future<Output = Option<Address>> + 'static {

        let lyquid = self.lyquid(&id);

        let bartender = self.bartender.clone();

        async move {

            let lyquid = lyquid?;

            let ln = lyquid.get_lyquid_number(pos).await.ok().flatten()?;

            Self::bartender_get_eth_addr(bartender, pos, id, ln.image).await

        }

    }

    async fn bartender_get_eth_addr(bartender: Lyquid, pos: ChainPos, id: LyquidID, ln_image: u32) -> Option<Address> {

        let call = {

            let bar_ln = bartender.get_lyquid_number(pos).await.ok().flatten()?;

            bartender

                .instance

                .call_instance_func_decoded(

                    bar_ln,

                    CallParams::builder()

                        .caller(Address::ZERO)

                        .method("get_eth_addr".into())

                        .input(encode_by_fields!(id: LyquidID = id, ln_image: u32 = ln_image).into())

                        .build(),

                    RunOptions::new(RunSource::Bartender),

                )

                .await

        };

        let result: Option<Address> = call.await.ok()?;

        result

    }

    fn get_address_by_ed25519(&self, pos: ChainPos, id: NodeID) -> impl Future<Output = Option<Address>> + 'static {

        let bartender = self.bartender.clone();

        async move {

            let call = {

                let bar_ln = bartender.get_lyquid_number(pos).await.ok().flatten()?;

                bartender

                    .instance

                    .call_instance_func_decoded(

                        bar_ln,

                        CallParams::builder()

                            .caller(Address::ZERO)

                            .method("get_address_by_ed25519".into())

                            .input(encode_by_fields!(id: NodeID).into())

                            .build(),

                        RunOptions::new(RunSource::Bartender),

                    )

                    .await

            };

            let result: Option<Address> = call.await.ok()?;

            result

        }

    }

    fn get_ed25519_by_address(

        &self, pos: ChainPos, address: Address,

    ) -> impl Future<Output = Option<NodeID>> + 'static {

        let bartender = self.bartender.clone();

        async move {

            let call = {

                let bar_ln = bartender.get_lyquid_number(pos).await.ok().flatten()?;

                bartender

                    .instance

                    .call_instance_func_decoded(

                        bar_ln,

                        CallParams::builder()

                            .caller(Address::ZERO)

                            .method("get_ed25519_by_address".into())

                            .input(encode_by_fields!(address: Address).into())

                            .build(),

                        RunOptions::new(RunSource::Bartender),

                    )

                    .await

            };

            let result: Option<NodeID> = call.await.ok()?;

            result

        }

    }

}