dbos 0.6.0

//! The connection to the system database, and the settings every call through it carries.
//!
//! **What a [`Client`](crate::Client) is, and what an [`Executor`](crate::Executor) has.** Every
//! DBOS operation that is not *running a workflow* is a read or a write against the system
//! database plus three things that decide how it is spelled: which serializer encodes the
//! payloads, which application owns the rows, and how often a wait re-asks. That bundle is this
//! type. An executor is this **plus** everything execution needs — an executor id and an
//! application version to stamp claims with, a runtime to spawn onto, a registry snapshot, task
//! sets to shut down.
//!
//! Splitting it this way is what lets a client hold no executor: the operations both surfaces need
//! take what they use rather than a whole executor, so a client calls them with no executor in
//! sight.
//!
//! # `Connection`, not `Database`
//!
//! [`SystemDatabase`] is already a type here — the trait a backend implements, which exists so a
//! second one can be dropped in without its callers changing. A second name a word away from it
//! would make every mention of "the database" a question about which layer is meant. What this
//! type adds to that trait object is not another database; it is *how this process talks to the one
//! it has*, which is what a connection is.
//!
//! # Where its methods live
//!
//! An operation both surfaces need is an inherent method here, **written in the module that owns
//! its types** rather than in this one: the five queue operations are in [`queue`](crate::queue),
//! reading an event is in [`event`](crate::event), and awaiting an outcome another execution owns
//! is in [`workflow`](crate::workflow). That is already how this crate spreads `impl DBOS`, and it
//! keeps this module to what a connection *is* while each operation stays next to the types it
//! speaks in.
//!
//! It is also what keeps [`WorkflowHandle`](crate::WorkflowHandle) one type. A handle needs the
//! connection, the serializer and the poll interval, and none of the executing half — so it holds
//! an `Arc<Connection>`, and a handle from a client and a handle from a running application are the
//! same thing rather than two types a caller has to tell apart.

use std::time::Duration;

use crate::checkpoint::StepWriter;
use crate::config::Config;
use crate::serialization::Serializer;
use crate::sysdb::types::StepCaller;
use crate::sysdb::{SystemDatabase, postgres};
use crate::{ClientConfig, Error, Result};

/// A connection to the system database, and the settings every call through it carries.
pub(crate) struct Connection {
    sysdb: Box<dyn SystemDatabase>,
    serializer: Serializer,
    app_name: Option<String>,
    outcome_poll_interval: Duration,
    owner: Owner,
}

/// Which of the two surfaces a connection was opened for.
///
/// The one thing a connection remembers about where it came from, and it is remembered for one
/// question: when a workflow awaits a handle whose connection is not its own, is that a mistake or
/// a plain wait? Two instances of an application disagreeing is the mistake — each has a step
/// counter the other cannot see, and the caller meant one of them. A client is not a second
/// instance of anything: it has no executor, no counter and nothing to checkpoint against, so a
/// workflow reaching through one is doing the only thing a client can do.
///
/// Nothing else asks, which is why this is a flag on the connection rather than a second type: a
/// client's connection and an application's differ in the settings above and in nothing a call
/// through them has to branch on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Owner {
    /// An [`Executor`](crate::Executor)'s: the connection an application runs its workflows
    /// through, with a step counter behind it.
    Application,
    /// A [`Client`](crate::Client)'s, which runs nothing.
    Client,
}

impl Connection {
    /// Connects for an application, migrating unless it was told not to.
    ///
    /// The executor's half of the connect: it takes an id, because an application's every
    /// statement is made by an identified process, and it may create and migrate the database
    /// because the application owns it.
    ///
    /// `app_name` is the **resolved** name — [`identity::resolve`](crate::identity::resolve)'s,
    /// not [`Config::app_name`](crate::Config::app_name)'s. On DBOS Cloud the deployment's
    /// `DBOS_APP_NAME` outranks whatever the application was built believing, and the name that
    /// stamps rows has to be the one that won.
    pub(crate) async fn for_application(
        config: &Config,
        executor_id: &str,
        app_name: &str,
    ) -> Result<Self> {
        let sysdb = postgres::PostgresSystemDatabase::connect(&postgres::Config {
            url: &config.database_url,
            max_connections: config.max_connections,
            use_listen_notify: config.use_listen_notify,
            idle_transaction_timeout: config.idle_transaction_timeout,
            migrate: config.migrate,
            settings: postgres::Settings {
                schema: &config.schema,
                executor_id: Some(executor_id),
                application_name: Some(app_name),
                polling_concurrency: config.polling_concurrency,
                notification_coalesce: config.notification_coalesce,
                ..postgres::Settings::default()
            },
        })
        .await
        .map_err(Error::SystemDatabase)?;

        Ok(Self {
            sysdb: Box::new(sysdb),
            serializer: config.serializer.clone(),
            app_name: Some(app_name.to_owned()),
            outcome_poll_interval: config.outcome_poll_interval(),
            owner: Owner::Application,
        })
    }

    /// Connects for a [`Client`](crate::Client), which owns nothing here.
    ///
    /// The differences from [`for_application`](Self::for_application) are the whole of what a
    /// client is:
    ///
    /// - **No executor id.** A client claims no workflow, so there is no process for the column to
    ///   name and no statement of its is made on one's behalf.
    /// - **It does not migrate**, and does not create the database either. A client is a guest:
    ///   the system database must already have been created by a DBOS application, and an
    ///   operator's tool that can rewrite the schema it is inspecting is a tool that can
    ///   break the application it was pointed at. There is no knob:
    ///   [`ClientConfig`] has no `migrate` field to set.
    ///   It does **verify**, because that is what this crate's `migrate: false` means:
    ///   a schema that is missing or behind what this build's queries are written against fails
    ///   here, naming the version it found, instead of surfacing as a confusing SQL error on the
    ///   first real call.
    /// - **The application name may be absent**, which no application's may be. A nameless client
    ///   writes unclaimed rows and reads across every application.
    /// - **Notifications are opt-in.** The listener and notifier start only when
    ///   [`ClientConfig::use_listen_notify`] is on, and it is off by default. A client can't tell
    ///   whether the application's executors push notifications, and a listening client's event
    ///   waits re-read only once a minute, so against executors that don't push they stall.
    ///   Polling every second is correct against any application.
    pub(crate) async fn for_client(config: &ClientConfig) -> Result<Self> {
        config.validate()?;

        let sysdb = postgres::PostgresSystemDatabase::connect(&postgres::Config {
            url: &config.database_url,
            max_connections: config.max_connections,
            use_listen_notify: config.use_listen_notify,
            idle_transaction_timeout: config.idle_transaction_timeout,
            migrate: false,
            settings: postgres::Settings {
                schema: &config.schema,
                executor_id: None,
                application_name: config.app_name.as_deref(),
                polling_concurrency: config.polling_concurrency,
                notification_coalesce: config.notification_coalesce,
                ..postgres::Settings::default()
            },
        })
        .await
        .map_err(Error::SystemDatabase)?;

        tracing::info!(
            app_name = config.app_name.as_deref().unwrap_or("<none>"),
            "DBOS client connected"
        );

        Ok(Self {
            sysdb: Box::new(sysdb),
            serializer: config.serializer.clone(),
            app_name: config.app_name.clone(),
            outcome_poll_interval: config.outcome_poll_interval(),
            owner: Owner::Client,
        })
    }

    /// The system database itself.
    pub(crate) fn sysdb(&self) -> &dyn SystemDatabase {
        &*self.sysdb
    }

    /// How payloads are encoded on their way in and out.
    pub(crate) fn serializer(&self) -> &Serializer {
        &self.serializer
    }

    /// The system database's view of a call recorded as a step: the ids, the calling execution's
    /// token, and this connection's serializer to record the call's result in.
    ///
    /// The result is a checkpoint only this SDK replays, so it is written in the configured
    /// serializer, as a step's output is.
    pub(crate) fn step_caller<'a>(
        &'a self,
        caller: Option<StepWriter<'a>>,
    ) -> Option<StepCaller<'a>> {
        caller.map(|caller| StepCaller {
            workflow_id: caller.workflow_id,
            step_id: caller.step_id,
            codec: &self.serializer,
            owner_xid: caller.owner_xid,
        })
    }

    /// The application whose rows this handle owns, or `None` for a nameless one.
    ///
    /// A nameless handle writes *unclaimed* rows, which belong to every application, and reads
    /// across all of them. Only a [`Client`](crate::Client) can be one:
    /// [`Config::app_name`](crate::Config::app_name) is required and validated, so an application
    /// always has a name.
    pub(crate) fn app_name(&self) -> Option<&str> {
        self.app_name.as_deref()
    }

    /// How often a caller waiting on a workflow it does not own asks whether it has finished.
    pub(crate) fn outcome_poll_interval(&self) -> Duration {
        self.outcome_poll_interval
    }

    /// Which surface this connection was opened for. See [`Owner`].
    pub(crate) fn owner(&self) -> Owner {
        self.owner
    }

    /// Releases the connections, the listener and the notifier.
    ///
    /// Idempotent, and on this type rather than on its holders because the connection is what is
    /// being closed: [`DBOS::shutdown`](crate::DBOS::shutdown) reaches it after stopping the
    /// workflows an executor was running, and [`Client::close`](crate::Client::close) reaches it
    /// with nothing to stop first.
    pub(crate) async fn close(&self) {
        self.sysdb.close().await;
    }
}

impl std::fmt::Debug for Connection {
    /// Hand-written because a `dyn SystemDatabase` is not `Debug` and should not become so: the
    /// trait exists to keep a driver type out of the engine, and a `Debug` bound would put one back
    /// in through the derive. The connection URL is deliberately not among the fields — see
    /// [`DBOS`](crate::DBOS)'s own `Debug`.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Connection")
            .field("app_name", &self.app_name)
            .field("serializer", &self.serializer)
            .finish_non_exhaustive()
    }
}