//! RPC connection support, mainloop, and protocol implementation.

pub(crate) mod auth;

mod methods;

use std::{

    collections::HashMap,

    io::Error as IoError,

    pin::Pin,

    sync::{Arc, Mutex, RwLock, Weak},

};

use asynchronous_codec::JsonCodecError;

use derive_deftly::Deftly;

use futures::{

    AsyncWriteExt as _, FutureExt, Sink, SinkExt as _, StreamExt,

    channel::mpsc,

    stream::{FusedStream, FuturesUnordered},

};

use rpc::dispatch::BoxedUpdateSink;

use serde_json::error::Category as JsonErrorCategory;

use tor_async_utils::{SinkExt as _, mpsc_channel_no_memquota};

use crate::{

    RpcAuthentication,

    cancel::{self, Cancel, CancelHandle},

    err::RequestParseError,

    globalid::{GlobalId, MacKey},

    msgs::{BoxedResponse, FlexibleRequest, ReqMeta, Request, RequestId, ResponseBody},

    objmap::{GenIdx, ObjMap},

};

use tor_rpcbase::templates::*;

use tor_rpcbase::{self as rpc, RpcError};

/// A function we use to construct Session objects in response to authentication.

//

// TODO RPC: Perhaps this should return a Result?

type SessionFactory = Box<dyn Fn(&RpcAuthentication) -> Arc<dyn rpc::Object> + Send + Sync>;

/// An open connection from an RPC client.

///

/// Tracks information that persists from one request to another.

///

/// The client might not have authenticated;

/// access and permissions control is handled via the capability system.

/// Specifically, the `objects` table in `Inner` hold capabilities

/// that the client will use to do things,

/// including an `RpcSession`.

///

/// # In the Arti RPC System

///

/// A connection to Arti.

///

/// This object is available as soon as you open a connection to Arti RPC,

/// even before you authenticate.  Its ObjectId is always `"connection"`.

///

/// Because this object is available before authentication,

/// it provides only those methods that you need

/// in order to perform authentication

/// and receive an `RpcSession`.

///

/// Note that a connection can only be authenticated once:

/// If you drop the `RpcSession` returned by authenticating,

/// you cannot get another one on the same connection.

#[derive(Deftly)]

#[derive_deftly(Object)]

pub struct Connection {

    /// The mutable state of this connection.

    inner: Mutex<Inner>,

    /// Lookup table to find the implementations for methods

    /// based on RPC object and method types.

    ///

    /// **NOTE: observe the [Lock hierarchy](crate::mgr::Inner#lock-hierarchy)**

    dispatch_table: Arc<RwLock<rpc::DispatchTable>>,

    /// A unique identifier for this connection.

    ///

    /// This kind of ID is used to refer to the connection from _outside_ of the

    /// context of an RPC connection: it can uniquely identify the connection

    /// from e.g. a SOCKS session so that clients can attach streams to it.

    connection_id: ConnectionId,

    /// A `MacKey` used to create `GlobalIds` for the objects whose identifiers

    /// need to exist outside this connection.

    global_id_mac_key: MacKey,

    /// The authentication type that's required in order to get a session.

    require_auth: tor_rpc_connect::auth::RpcAuth,

}

/// The inner, lock-protected part of an RPC connection.

struct Inner {

    /// Map from request ID to handles; used when we need to cancel a request.

    //

    // TODO: We have two options here for handling colliding IDs.  We can either turn

    // this into a multimap, or we can declare that cancelling a request only

    // cancels the most recent request sent with that ID.

    inflight: HashMap<RequestId, Option<CancelHandle>>,

    /// An object map used to look up most objects by ID, and keep track of

    /// which objects are owned by this connection.

    objects: ObjMap,

    /// A reference to this connection itself.

    ///

    /// Used when we're looking up the connection within the RPC system as an object.

    ///

    /// TODO RPC: Maybe there is an easier way to do this while keeping `context` object-save?

    this_connection: Option<Weak<Connection>>,

    /// A SessionFactory that will be used to create a session if authentication is successful.

    ///

    /// This is None if the connection has already been authenticated.

    session_factory: Option<SessionFactory>,

}

/// How many updates can be pending, per connection, before they start to block?

const UPDATE_CHAN_SIZE: usize = 128;

/// A type-erased [`FusedStream`] yielding [`Request`]s.

//

// (We name this type and [`BoxedResponseSink`] below so as to keep the signature for run_loop

// nice and simple.)

pub(crate) type BoxedRequestStream = Pin<

    Box<dyn FusedStream<Item = Result<FlexibleRequest, asynchronous_codec::JsonCodecError>> + Send>,

>;

/// A type-erased [`Sink`] accepting [`BoxedResponse`]s.

pub(crate) type BoxedResponseSink =

    Pin<Box<dyn Sink<BoxedResponse, Error = asynchronous_codec::JsonCodecError> + Send>>;

/// A random value used to identify an connection.

#[derive(

    Copy,

    Clone,

    Debug,

    Eq,

    PartialEq,

    Hash,

    derive_more::From,

    derive_more::Into,

    derive_more::AsRef,

)]

pub(crate) struct ConnectionId([u8; 16]);

impl ConnectionId {

    /// The length of a ConnectionId.

    pub(crate) const LEN: usize = 16;

}

impl Connection {

    /// A special object ID that indicates the connection itself.

    ///

    /// On a fresh connection, this is the only ObjectId that exists.

    //

    // TODO: We might want to move responsibility for tracking this ID and its value into ObjMap.

    const CONNECTION_OBJ_ID: &'static str = "connection";

    /// Create a new connection.

    /// Construct a new object to serve as the `session` for a connection.

            )

    /// If possible, convert an `ObjectId` into a `GenIdx` that can be used in

    /// this connection's ObjMap.

        // Design note: It's not really necessary from a security POV to

        // check the MAC here; any possible GenIdx we return will either

        // refer to some object we're allowed to name in this session, or to

        // no object at all.  Still, we check anyway, since it shouldn't

        // hurt to do so.

            // We have a GlobalId with a valid MAC. Let's make sure it applies

            // to this connection's ObjMap.  (We do not support referring to

            // anyone else's objects.)

            //

            // Design note: As above, this check is a protection against

            // accidental misuse, not a security feature: even if we removed

            // this check, we would still only allow objects that this session

            // is allowed to name.

            } else {

            }

        } else {

            // It's not a GlobalId; let's see if we can make sense of it as an

            // ObjMap index.

        }

    /// Look up a given object by its object ID relative to this connection.

        } else {

        }

    /// As `lookup_object`, but expect a `GenIdx`.

    /// Un-register the request `id` and stop tracking its information.

    /// Register the request `id` as a cancellable request.

    ///

    /// If `handle` is none, register it as an uncancellable request.

    /// Try to cancel the request `id`.

    ///

    /// Return an error when `id` cannot be found, or cannot be cancelled.

    /// (These cases are indistinguishable.)

            }

            Some(None) => {

                // Put it back in case somebody tries again.

            }

        }

    /// Run in a loop, decoding JSON requests from `input` and

    /// writing JSON responses onto `output`.

        /// Banner line to send, indicating that Arti is ready to receive requests.

        ///

        /// The key in this json object is mandatory; the value can be anything.

        const BANNER: &[u8] = b"{\"arti_rpc\":{}}\n";

        ));

            )

        );

    /// Run in a loop, handling requests from `request_stream` and writing

    /// responses onto `response_stream`.

    ///

    /// After this returns, even if it returns `Ok(())`, the connection must no longer be used.

        // This function will multiplex on three streams:

        // * `request_stream` -- a stream of incoming requests from the client.

        // * `finished_requests` -- a stream of requests that are done.

        // * `rx_response` -- a stream of updates and final responses sent from

        //   in-progress tasks. (We put updates and final responsese onto the

        //   same channel to ensure that they stay in-order for each method

        //   invocation.

        //

        // Note that the blocking behavior here is deliberate: We want _all_ of

        // these reads to start blocking when response_sink.send is blocked.

        // TODO RPC should this queue participate in memquota?

        /// Helper: enforce an explicit "continue".

        struct Continue;

        // We create a separate async block here and immediately await it,

        // so that any internal `returns` and `?`s do not escape the function.

            loop {

                        // A task is done, so we can forget about it.

                    }

                        // The future for some request has sent a response (success,

                        // failure, or update), so we can inform the client.

                        // Calling `await` here (and below) is deliberate: we _want_

                        // to stop reading the client's requests if the client is

                        // not reading their responses (or not) reading them fast

                        // enough.

                    }

                            None => {

                                // We've reached the end of the stream of requests;

                                // time to close.

                            }

                                // We got a non-recoverable error from the JSON codec.

                            }

                                // We decoded the request as Json, but not as a `Valid`` request.

                                // Send back a response indicating what was wrong with it.

                                );

                                    // The spec says we must close the connection in this case.

                            }

                                // We have a request. Time to launch it!

                            }

                        }

                    }

                };

            }

        }

    /// Invoke `request` and send all of its responses to `tx_response`.

        let Request {

        } else {

        };

        // Create `run_method_lowlevel` future, and make it cancellable.

        // Optionally register the future as cancellable.  Then run it to completion.

        } else {

        };

        // Figure out how to respond.

            // TODO: If we're going to box this, let's do so earlier.

                        err

                    );

            }

        };

        // Send the response.

        //

        // (It's okay to ignore the error here, since it can only mean that the

        // RPC connection has closed.)

        // Unregister the request.

        //

        // TODO: This may unregister a different request if the user sent

        // in another request with the same ID.

    /// Run a single method, and return its final response.

    ///

    /// If `tx_updates` is provided, and this method generates updates, it

    /// should send those updates on `tx_updates`

    ///

    /// Note that this function is able to send responses with IDs that do not

    /// match the original.  It should enforce correct IDs on whatever response

    /// it generates.

            // TODO RPC: Eventually, we will need a way to tell which "features" are actually

            // available.  But for now, we have no features, so if the require list is nonempty,

            // we can safely reject the request.

        // Note that we drop the read lock before we await this future!

}

/// An error returned when an RPC request lists some feature as required,

/// but we don't have every such feature.

#[derive(Clone, Debug, thiserror::Error)]

#[error("Required features not available")]

struct MissingFeaturesError(

    /// A list of the features that were requested but not available.

    Vec<String>,

);

impl From<MissingFeaturesError> for RpcError {

        );

}

/// A failure that results in closing a [`Connection`].

#[derive(Clone, Debug, thiserror::Error)]

#[non_exhaustive]

pub enum ConnectionError {

    /// Unable to write to our connection.

    #[error("Could not write to connection")]

    WriteFailed(#[source] Arc<IoError>),

    /// Read error from connection.

    #[error("Problem reading from connection")]

    ReadFailed(#[source] Arc<IoError>),

    /// Read something that we could not decode.

    #[error("Unable to decode request from connection")]

    DecodeFailed(#[source] Arc<serde_json::Error>),

    /// Unable to write our response as json.

    #[error("Unable to encode response onto connection")]

    EncodeFailed(#[source] Arc<serde_json::Error>),

    /// We encountered a problem when parsing a request that was (in our judgment)

    /// too severe to recover from.

    #[error("Unrecoverable problem from parsed request")]

    RequestParseFailed(#[from] RequestParseError),

}

impl ConnectionError {

    /// Construct a new `ConnectionError` from a `JsonCodecError` that has occurred while writing.

        }

    /// Return true if this error is (or might be) due to the peer closing the connection.

    ///

    /// Such errors should be tolerated without much complaint;

    /// other errors should at least be logged somewhere.

        use JsonErrorCategory as JK;

        use std::io::ErrorKind as IK;

        #[allow(clippy::match_like_matches_macro)]

            },

            },

        }

    /// Construct a `ConnectionError` from a JsonCodecError that occurred while reading.

        }

}

impl rpc::Context for Connection {

        // Design note: It is a deliberate decision to _always_ use GlobalId for

        // objects whose IDs are _ever_ exported for use in SOCKS requests.  Some

        // alternatives would be to use GlobalId conditionally, or to have a

        // separate Method to create a new GlobalId given an existing LocalId.

        } else {

        }

        } else {

        };

        } else {

        }

}

/// An error given when an RPC request is cancelled.

///

/// This is a separate type from [`crate::cancel::Cancelled`] since eventually

/// we want to move that type into a general-purpose location, and make it not

/// RPC-specific.

#[derive(thiserror::Error, Clone, Debug, serde::Serialize)]

#[error("RPC request was cancelled")]

pub(crate) struct RequestCancelled;

impl From<RequestCancelled> for RpcError {

        )

}

/// An error given when we attempt to cancel an RPC request, but cannot.

///

#[derive(thiserror::Error, Clone, Debug, serde::Serialize)]

pub(crate) enum CancelError {

    /// We didn't find any request with the provided ID.

    ///

    /// Since we don't keep track of requests after they finish or are cancelled,

    /// we cannot distinguish the cases where a request has finished,

    /// where the request has been cancelled,

    /// or where the request never existed.

    /// Therefore we collapse them into a single error type.

    #[error("RPC request not found")]

    RequestNotFound,

    /// This kind of request cannot be cancelled.

    #[error("Uncancellable request")]

    CannotCancelRequest,

    /// We tried to cancel a request but found out it was already cancelled.

    ///

    /// This error should be impossible.

    #[error("Request somehow cancelled twice!")]

    AlreadyCancelled,

}

impl From<cancel::CannotCancel> for CancelError {

        use CancelError as CE;

        use cancel::CannotCancel as CC;

            // We map "finished" to RequestNotFound since it is not in the general case

            // distinguishable from it; see documentation on RequestNotFound.

        }

}

impl From<CancelError> for RpcError {

        use CancelError as CE;

        use rpc::RpcErrorKind as REK;

        };

}