{-# LANGUAGE UndecidableInstances #-}

-- | Streaming-mode @COPY@: the data flows through the client connection
-- rather than to or from a file on the database server.
--
-- For the file-mode variant, see
-- "Database.Beam.Backend.SQL.BeamExtensions.Copy.File".
module Database.Beam.Backend.SQL.BeamExtensions.Copy.Stream
  ( -- * Building a streaming COPY statement
    copyTableToStream,
    copySelectToStream,
    SqlCopyToStream (..),
    copyTableFromStream,
    SqlCopyFromStream (..),

    -- * Streaming statement-level syntax classes
    IsSqlCopyToStreamSyntax (..),
    IsSqlCopyFromStreamSyntax (..),
    BeamSqlBackendCopyToStreamSyntax,
    BeamSqlBackendCopyFromStreamSyntax,

    -- * Runner classes
    MonadBeamCopyToStream (..),
    MonadBeamCopyFromStream (..),
  )
where

import Control.Monad.Cont (ContT)
import Control.Monad.Except (ExceptT)
import qualified Control.Monad.RWS.Lazy as Lazy
import qualified Control.Monad.RWS.Strict as Strict
import Control.Monad.Reader (ReaderT)
import qualified Control.Monad.State.Lazy as Lazy
import qualified Control.Monad.State.Strict as Strict
import Control.Monad.Trans (lift)
import qualified Control.Monad.Writer.Lazy as Lazy
import qualified Control.Monad.Writer.Strict as Strict
import Data.ByteString (ByteString)
import Data.Kind (Type)
import Data.List.NonEmpty (nonEmpty)
import Data.Text (Text)
import Database.Beam.Backend.SQL (BeamSqlBackendSelectSyntax, MonadBeam)
import Database.Beam.Backend.SQL.BeamExtensions.Copy.File
  ( IsSqlCopyFromSourceSyntax (..),
    IsSqlCopyToSourceSyntax (..),
    projection,
  )
import Database.Beam.Query (QField, SqlSelect (..))
import Database.Beam.Query.Internal (AnyType, ProjectibleWithPredicate)
import Database.Beam.Schema (TableEntity)
import Database.Beam.Schema.Tables
  ( DatabaseEntity (..),
    DatabaseEntityDescriptor (DatabaseTable),
    IsDatabaseEntity (dbEntityName, dbEntitySchema),
  )
import Lens.Micro ((^.))

-- | Statement-level syntax for backends that support streaming
-- @COPY ... TO@ (data leaving the database through the client connection).
--
-- Mirrors 'Database.Beam.Backend.SQL.BeamExtensions.Copy.File.IsSqlCopyToSyntax',
-- but the params value carries only format options — no destination path,
-- since chunks travel through the wire.
class (IsSqlCopyToSourceSyntax (SqlCopyToStreamSourceSyntax cmd)) => IsSqlCopyToStreamSyntax cmd where
  -- | The syntax for the source of the streaming copy. As with file-mode
  --  COPY, this can be a table (perhaps with a projection) or a @SELECT@
  --  query.
  type SqlCopyToStreamSourceSyntax cmd :: Type

  -- | All backend-specific options which determine HOW the streaming copy
  --  is performed (format, delimiter, etc.). Unlike the file-mode params,
  --  there is no destination — chunks flow through the connection.
  type SqlCopyToStreamParams cmd :: Type

  -- | Combine a source and a parameters value into a complete
  -- streaming @COPY ... TO@ stream statement.
  copyToStreamStmt ::
    SqlCopyToStreamSourceSyntax cmd ->
    SqlCopyToStreamParams cmd ->
    cmd

-- | Statement-level syntax for backends that support streaming
-- @COPY ... FROM@ (data entering the database through the client connection).
--
-- Mirrors 'Database.Beam.Backend.SQL.BeamExtensions.Copy.File.IsSqlCopyFromSyntax'.
class (IsSqlCopyFromSourceSyntax (SqlCopyFromStreamSourceSyntax cmd)) => IsSqlCopyFromStreamSyntax cmd where
  -- | The syntax for the destination table of the streaming copy.
  type SqlCopyFromStreamSourceSyntax cmd :: Type

  -- | All backend-specific options which determine HOW the streaming copy
  --  is performed. Unlike the file-mode params, there is no source path —
  --  chunks flow through the connection.
  type SqlCopyFromStreamParams cmd :: Type

  -- | Combine a destination and a parameters value into a complete
  -- streaming @COPY ... FROM@ stream statement.
  copyFromStreamStmt ::
    SqlCopyFromStreamSourceSyntax cmd ->
    SqlCopyFromStreamParams cmd ->
    cmd

-- | Type-family selector for the backend's streaming @COPY ... TO@ syntax.
-- A backend instance binds this to the concrete syntax type that
-- implements 'IsSqlCopyToStreamSyntax'.
type family BeamSqlBackendCopyToStreamSyntax be :: Type

-- | Type-family selector for the backend's streaming @COPY ... FROM@ syntax.
-- See 'BeamSqlBackendCopyToStreamSyntax'.
type family BeamSqlBackendCopyFromStreamSyntax be :: Type

-- | A built streaming-mode @COPY ... TO@ statement, ready to be executed by
-- 'runCopyToStream'.
data SqlCopyToStream be a
  = SqlCopyToStream !(BeamSqlBackendCopyToStreamSyntax be)
  | -- | A projection covering zero columns. 'runCopyToStream' should treat
    --    this as a no-op (it must still call the sink zero times) rather
    --    than emit an empty @COPY tbl () TO STDOUT@ statement.
    SqlCopyToStreamNoColumns

-- | A built streaming-mode @COPY ... FROM@ statement, ready to be executed
-- by 'runCopyFromStream'.
data SqlCopyFromStream be a
  = SqlCopyFromStream !(BeamSqlBackendCopyFromStreamSyntax be)
  | -- | A projection covering zero columns. 'runCopyFromStream' should
    --    treat this as a no-op (it should not pull from the source) rather
    --    than emit an empty @COPY tbl () FROM STDIN@ statement.
    SqlCopyFromStreamNoColumns

-- | Express a streaming copy from a table, the data flowing through the
-- client connection rather than to a server-side file.
--
-- To stream the result of a @SELECT@ query instead, see 'copySelectToStream'.
--
-- @since 0.11.1.0
copyTableToStream ::
  ( IsSqlCopyToStreamSyntax (BeamSqlBackendCopyToStreamSyntax be),
    ProjectibleWithPredicate AnyType () Text proj
  ) =>
  DatabaseEntity be db (TableEntity table) ->
  -- | Projection from table to columns. If you want
  --  to copy the entire table, use 'id'.
  (table (QField s) -> proj) ->
  -- | Backend-specific options.
  SqlCopyToStreamParams (BeamSqlBackendCopyToStreamSyntax be) ->
  SqlCopyToStream be proj
copyTableToStream :: forall be proj (db :: (* -> *) -> *) (table :: (* -> *) -> *) s.
(IsSqlCopyToStreamSyntax (BeamSqlBackendCopyToStreamSyntax be),
 ProjectibleWithPredicate AnyType () Text proj) =>
DatabaseEntity be db (TableEntity table)
-> (table (QField s) -> proj)
-> SqlCopyToStreamParams (BeamSqlBackendCopyToStreamSyntax be)
-> SqlCopyToStream be proj
copyTableToStream (DatabaseEntity dt :: DatabaseEntityDescriptor be (TableEntity table)
dt@(DatabaseTable {})) table (QField s) -> proj
mkProj SqlCopyToStreamParams (BeamSqlBackendCopyToStreamSyntax be)
options =
  case [Text] -> Maybe (NonEmpty Text)
forall a. [a] -> Maybe (NonEmpty a)
nonEmpty (DatabaseEntityDescriptor be (TableEntity table)
-> (table (QField s) -> proj) -> [Text]
forall proj (table :: (* -> *) -> *) be s.
(ProjectibleWithPredicate AnyType () Text proj, Beamable table) =>
DatabaseEntityDescriptor be (TableEntity table)
-> (table (QField s) -> proj) -> [Text]
projection DatabaseEntityDescriptor be (TableEntity table)
dt table (QField s) -> proj
mkProj) of
    Maybe (NonEmpty Text)
Nothing -> SqlCopyToStream be proj
forall be a. SqlCopyToStream be a
SqlCopyToStreamNoColumns
    Just NonEmpty Text
cols ->
      let source :: SqlCopyToStreamSourceSyntax (BeamSqlBackendCopyToStreamSyntax be)
source = Maybe Text
-> Text
-> Maybe (NonEmpty Text)
-> SqlCopyToStreamSourceSyntax
     (BeamSqlBackendCopyToStreamSyntax be)
forall syntax.
IsSqlCopyToSourceSyntax syntax =>
Maybe Text -> Text -> Maybe (NonEmpty Text) -> syntax
copyTableToSyntax (DatabaseEntityDescriptor be (TableEntity table)
dt DatabaseEntityDescriptor be (TableEntity table)
-> Getting
     (Maybe Text)
     (DatabaseEntityDescriptor be (TableEntity table))
     (Maybe Text)
-> Maybe Text
forall s a. s -> Getting a s a -> a
^. Getting
  (Maybe Text)
  (DatabaseEntityDescriptor be (TableEntity table))
  (Maybe Text)
forall be entityType.
IsDatabaseEntity be entityType =>
Traversal' (DatabaseEntityDescriptor be entityType) (Maybe Text)
Traversal'
  (DatabaseEntityDescriptor be (TableEntity table)) (Maybe Text)
dbEntitySchema) (DatabaseEntityDescriptor be (TableEntity table)
dt DatabaseEntityDescriptor be (TableEntity table)
-> Getting
     Text (DatabaseEntityDescriptor be (TableEntity table)) Text
-> Text
forall s a. s -> Getting a s a -> a
^. Getting Text (DatabaseEntityDescriptor be (TableEntity table)) Text
forall be entityType.
IsDatabaseEntity be entityType =>
Lens' (DatabaseEntityDescriptor be entityType) Text
Lens' (DatabaseEntityDescriptor be (TableEntity table)) Text
dbEntityName) (NonEmpty Text -> Maybe (NonEmpty Text)
forall a. a -> Maybe a
Just NonEmpty Text
cols)
       in BeamSqlBackendCopyToStreamSyntax be -> SqlCopyToStream be proj
forall be a.
BeamSqlBackendCopyToStreamSyntax be -> SqlCopyToStream be a
SqlCopyToStream
            (SqlCopyToStreamSourceSyntax (BeamSqlBackendCopyToStreamSyntax be)
-> SqlCopyToStreamParams (BeamSqlBackendCopyToStreamSyntax be)
-> BeamSqlBackendCopyToStreamSyntax be
forall cmd.
IsSqlCopyToStreamSyntax cmd =>
SqlCopyToStreamSourceSyntax cmd -> SqlCopyToStreamParams cmd -> cmd
copyToStreamStmt SqlCopyToStreamSourceSyntax (BeamSqlBackendCopyToStreamSyntax be)
source SqlCopyToStreamParams (BeamSqlBackendCopyToStreamSyntax be)
options)

-- | Express a streaming copy from the result of a @SELECT@ statement.
--
-- To stream a table, or a subset of columns, see 'copyTableToStream'.
--
-- @since 0.11.1.0
copySelectToStream ::
  ( IsSqlCopyToStreamSyntax (BeamSqlBackendCopyToStreamSyntax be),
    SqlCopyToSourceSelectSyntax (SqlCopyToStreamSourceSyntax (BeamSqlBackendCopyToStreamSyntax be))
      ~ BeamSqlBackendSelectSyntax be
  ) =>
  SqlSelect be a ->
  SqlCopyToStreamParams (BeamSqlBackendCopyToStreamSyntax be) ->
  SqlCopyToStream be a
copySelectToStream :: forall be a.
(IsSqlCopyToStreamSyntax (BeamSqlBackendCopyToStreamSyntax be),
 SqlCopyToSourceSelectSyntax
   (SqlCopyToStreamSourceSyntax (BeamSqlBackendCopyToStreamSyntax be))
 ~ BeamSqlBackendSelectSyntax be) =>
SqlSelect be a
-> SqlCopyToStreamParams (BeamSqlBackendCopyToStreamSyntax be)
-> SqlCopyToStream be a
copySelectToStream (SqlSelect BeamSqlBackendSelectSyntax be
selectSyntax) SqlCopyToStreamParams (BeamSqlBackendCopyToStreamSyntax be)
options =
  let source :: SqlCopyToStreamSourceSyntax (BeamSqlBackendCopyToStreamSyntax be)
source = SqlCopyToSourceSelectSyntax
  (SqlCopyToStreamSourceSyntax (BeamSqlBackendCopyToStreamSyntax be))
-> SqlCopyToStreamSourceSyntax
     (BeamSqlBackendCopyToStreamSyntax be)
forall syntax.
IsSqlCopyToSourceSyntax syntax =>
SqlCopyToSourceSelectSyntax syntax -> syntax
copySelectToSyntax BeamSqlBackendSelectSyntax be
SqlCopyToSourceSelectSyntax
  (SqlCopyToStreamSourceSyntax (BeamSqlBackendCopyToStreamSyntax be))
selectSyntax
   in BeamSqlBackendCopyToStreamSyntax be -> SqlCopyToStream be a
forall be a.
BeamSqlBackendCopyToStreamSyntax be -> SqlCopyToStream be a
SqlCopyToStream (SqlCopyToStreamSourceSyntax (BeamSqlBackendCopyToStreamSyntax be)
-> SqlCopyToStreamParams (BeamSqlBackendCopyToStreamSyntax be)
-> BeamSqlBackendCopyToStreamSyntax be
forall cmd.
IsSqlCopyToStreamSyntax cmd =>
SqlCopyToStreamSourceSyntax cmd -> SqlCopyToStreamParams cmd -> cmd
copyToStreamStmt SqlCopyToStreamSourceSyntax (BeamSqlBackendCopyToStreamSyntax be)
source SqlCopyToStreamParams (BeamSqlBackendCopyToStreamSyntax be)
options)

-- | Express a streaming copy into a table, the data flowing through the
-- client connection rather than from a server-side file.
--
-- @since 0.11.1.0
copyTableFromStream ::
  ( IsSqlCopyFromStreamSyntax (BeamSqlBackendCopyFromStreamSyntax be),
    ProjectibleWithPredicate AnyType () Text proj
  ) =>
  DatabaseEntity be db (TableEntity table) ->
  -- | Projection from which to copy columns. Other columns
  -- will have their default value inserted.
  --
  -- To copy the stream into the entire table, use 'id'.
  (table (QField s) -> proj) ->
  SqlCopyFromStreamParams (BeamSqlBackendCopyFromStreamSyntax be) ->
  SqlCopyFromStream be proj
copyTableFromStream :: forall be proj (db :: (* -> *) -> *) (table :: (* -> *) -> *) s.
(IsSqlCopyFromStreamSyntax (BeamSqlBackendCopyFromStreamSyntax be),
 ProjectibleWithPredicate AnyType () Text proj) =>
DatabaseEntity be db (TableEntity table)
-> (table (QField s) -> proj)
-> SqlCopyFromStreamParams (BeamSqlBackendCopyFromStreamSyntax be)
-> SqlCopyFromStream be proj
copyTableFromStream (DatabaseEntity dt :: DatabaseEntityDescriptor be (TableEntity table)
dt@(DatabaseTable {})) table (QField s) -> proj
mkProj SqlCopyFromStreamParams (BeamSqlBackendCopyFromStreamSyntax be)
options =
  case [Text] -> Maybe (NonEmpty Text)
forall a. [a] -> Maybe (NonEmpty a)
nonEmpty (DatabaseEntityDescriptor be (TableEntity table)
-> (table (QField s) -> proj) -> [Text]
forall proj (table :: (* -> *) -> *) be s.
(ProjectibleWithPredicate AnyType () Text proj, Beamable table) =>
DatabaseEntityDescriptor be (TableEntity table)
-> (table (QField s) -> proj) -> [Text]
projection DatabaseEntityDescriptor be (TableEntity table)
dt table (QField s) -> proj
mkProj) of
    Maybe (NonEmpty Text)
Nothing -> SqlCopyFromStream be proj
forall be a. SqlCopyFromStream be a
SqlCopyFromStreamNoColumns
    Just NonEmpty Text
cols ->
      let source :: SqlCopyFromStreamSourceSyntax
  (BeamSqlBackendCopyFromStreamSyntax be)
source = Maybe Text
-> Text
-> Maybe (NonEmpty Text)
-> SqlCopyFromStreamSourceSyntax
     (BeamSqlBackendCopyFromStreamSyntax be)
forall syntax.
IsSqlCopyFromSourceSyntax syntax =>
Maybe Text -> Text -> Maybe (NonEmpty Text) -> syntax
copyTableFromSyntax (DatabaseEntityDescriptor be (TableEntity table)
dt DatabaseEntityDescriptor be (TableEntity table)
-> Getting
     (Maybe Text)
     (DatabaseEntityDescriptor be (TableEntity table))
     (Maybe Text)
-> Maybe Text
forall s a. s -> Getting a s a -> a
^. Getting
  (Maybe Text)
  (DatabaseEntityDescriptor be (TableEntity table))
  (Maybe Text)
forall be entityType.
IsDatabaseEntity be entityType =>
Traversal' (DatabaseEntityDescriptor be entityType) (Maybe Text)
Traversal'
  (DatabaseEntityDescriptor be (TableEntity table)) (Maybe Text)
dbEntitySchema) (DatabaseEntityDescriptor be (TableEntity table)
dt DatabaseEntityDescriptor be (TableEntity table)
-> Getting
     Text (DatabaseEntityDescriptor be (TableEntity table)) Text
-> Text
forall s a. s -> Getting a s a -> a
^. Getting Text (DatabaseEntityDescriptor be (TableEntity table)) Text
forall be entityType.
IsDatabaseEntity be entityType =>
Lens' (DatabaseEntityDescriptor be entityType) Text
Lens' (DatabaseEntityDescriptor be (TableEntity table)) Text
dbEntityName) (NonEmpty Text -> Maybe (NonEmpty Text)
forall a. a -> Maybe a
Just NonEmpty Text
cols)
       in BeamSqlBackendCopyFromStreamSyntax be -> SqlCopyFromStream be proj
forall be a.
BeamSqlBackendCopyFromStreamSyntax be -> SqlCopyFromStream be a
SqlCopyFromStream
            (SqlCopyFromStreamSourceSyntax
  (BeamSqlBackendCopyFromStreamSyntax be)
-> SqlCopyFromStreamParams (BeamSqlBackendCopyFromStreamSyntax be)
-> BeamSqlBackendCopyFromStreamSyntax be
forall cmd.
IsSqlCopyFromStreamSyntax cmd =>
SqlCopyFromStreamSourceSyntax cmd
-> SqlCopyFromStreamParams cmd -> cmd
copyFromStreamStmt SqlCopyFromStreamSourceSyntax
  (BeamSqlBackendCopyFromStreamSyntax be)
source SqlCopyFromStreamParams (BeamSqlBackendCopyFromStreamSyntax be)
options)

-- | 'MonadBeam's that support streaming data out of a database through the
-- client connection (e.g. PostgreSQL's @COPY ... TO STDOUT@).
--
-- The supplied @ByteString -> IO ()@ callback is invoked once per chunk
-- received. The 'runCopyToStream' call blocks until the COPY completes; on
-- failure it raises the underlying backend's exception.
--
-- See 'MonadBeamCopyFromStream' for the inverse operation.
--
-- @since 0.11.1.0
class (MonadBeam be m) => MonadBeamCopyToStream be m | m -> be where
  -- | Execute a built streaming @COPY ... TO@ stream statement. The supplied sink
  -- is invoked from 'IO' once per chunk emitted by the server, in order;
  -- 'runCopyToStream' returns once the server signals end of stream.
  runCopyToStream ::
    SqlCopyToStream be a ->
    -- | Sink. Called once for each chunk of bytes the server emits.
    (ByteString -> IO ()) ->
    m ()

instance (MonadBeamCopyToStream be m) => MonadBeamCopyToStream be (ExceptT e m) where
  runCopyToStream :: forall a.
SqlCopyToStream be a -> (ByteString -> IO ()) -> ExceptT e m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb = m () -> ExceptT e m ()
forall (m :: * -> *) a. Monad m => m a -> ExceptT e m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall a. SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyToStream be m =>
SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb)

instance (MonadBeamCopyToStream be m) => MonadBeamCopyToStream be (ContT r m) where
  runCopyToStream :: forall a.
SqlCopyToStream be a -> (ByteString -> IO ()) -> ContT r m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb = m () -> ContT r m ()
forall (m :: * -> *) a. Monad m => m a -> ContT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall a. SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyToStream be m =>
SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb)

instance (MonadBeamCopyToStream be m) => MonadBeamCopyToStream be (ReaderT r m) where
  runCopyToStream :: forall a.
SqlCopyToStream be a -> (ByteString -> IO ()) -> ReaderT r m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb = m () -> ReaderT r m ()
forall (m :: * -> *) a. Monad m => m a -> ReaderT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall a. SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyToStream be m =>
SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb)

instance (MonadBeamCopyToStream be m) => MonadBeamCopyToStream be (Lazy.StateT r m) where
  runCopyToStream :: forall a.
SqlCopyToStream be a -> (ByteString -> IO ()) -> StateT r m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb = m () -> StateT r m ()
forall (m :: * -> *) a. Monad m => m a -> StateT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall a. SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyToStream be m =>
SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb)

instance (MonadBeamCopyToStream be m) => MonadBeamCopyToStream be (Strict.StateT r m) where
  runCopyToStream :: forall a.
SqlCopyToStream be a -> (ByteString -> IO ()) -> StateT r m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb = m () -> StateT r m ()
forall (m :: * -> *) a. Monad m => m a -> StateT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall a. SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyToStream be m =>
SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb)

instance (MonadBeamCopyToStream be m, Monoid r) => MonadBeamCopyToStream be (Lazy.WriterT r m) where
  runCopyToStream :: forall a.
SqlCopyToStream be a -> (ByteString -> IO ()) -> WriterT r m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb = m () -> WriterT r m ()
forall (m :: * -> *) a. Monad m => m a -> WriterT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall a. SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyToStream be m =>
SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb)

instance (MonadBeamCopyToStream be m, Monoid r) => MonadBeamCopyToStream be (Strict.WriterT r m) where
  runCopyToStream :: forall a.
SqlCopyToStream be a -> (ByteString -> IO ()) -> WriterT r m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb = m () -> WriterT r m ()
forall (m :: * -> *) a. Monad m => m a -> WriterT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall a. SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyToStream be m =>
SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb)

instance (MonadBeamCopyToStream be m, Monoid w) => MonadBeamCopyToStream be (Lazy.RWST r w s m) where
  runCopyToStream :: forall a.
SqlCopyToStream be a -> (ByteString -> IO ()) -> RWST r w s m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb = m () -> RWST r w s m ()
forall (m :: * -> *) a. Monad m => m a -> RWST r w s m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall a. SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyToStream be m =>
SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb)

instance (MonadBeamCopyToStream be m, Monoid w) => MonadBeamCopyToStream be (Strict.RWST r w s m) where
  runCopyToStream :: forall a.
SqlCopyToStream be a -> (ByteString -> IO ()) -> RWST r w s m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb = m () -> RWST r w s m ()
forall (m :: * -> *) a. Monad m => m a -> RWST r w s m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall a. SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyToStream be m =>
SqlCopyToStream be a -> (ByteString -> IO ()) -> m ()
runCopyToStream SqlCopyToStream be a
s ByteString -> IO ()
cb)

-- | 'MonadBeam's that support streaming data into a database through the
-- client connection (e.g. PostgreSQL's @COPY ... FROM STDIN@).
--
-- The supplied @IO (Maybe ByteString)@ source is pulled repeatedly until it
-- returns 'Nothing', signalling end of data. 'runCopyFromStream' blocks
-- until the COPY commits; on failure it raises the underlying backend's
-- exception.
--
-- See 'MonadBeamCopyToStream' for the inverse operation.
--
-- @since 0.11.1.0
class (MonadBeam be m) => MonadBeamCopyFromStream be m | m -> be where
  -- | Execute a built streaming @COPY ... FROM@ statement. The supplied
  -- producer is pulled from 'IO' until it returns 'Nothing'; each 'Just'
  -- chunk is forwarded to the server in order. 'runCopyFromStream' returns
  -- once the server has acknowledged the end of stream.
  runCopyFromStream ::
    SqlCopyFromStream be a ->
    -- | Source. Called repeatedly. 'Nothing' signals end of data.
    IO (Maybe ByteString) ->
    m ()

instance (MonadBeamCopyFromStream be m) => MonadBeamCopyFromStream be (ExceptT e m) where
  runCopyFromStream :: forall a.
SqlCopyFromStream be a -> IO (Maybe ByteString) -> ExceptT e m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer = m () -> ExceptT e m ()
forall (m :: * -> *) a. Monad m => m a -> ExceptT e m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall a. SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyFromStream be m =>
SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer)

instance (MonadBeamCopyFromStream be m) => MonadBeamCopyFromStream be (ContT r m) where
  runCopyFromStream :: forall a.
SqlCopyFromStream be a -> IO (Maybe ByteString) -> ContT r m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer = m () -> ContT r m ()
forall (m :: * -> *) a. Monad m => m a -> ContT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall a. SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyFromStream be m =>
SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer)

instance (MonadBeamCopyFromStream be m) => MonadBeamCopyFromStream be (ReaderT r m) where
  runCopyFromStream :: forall a.
SqlCopyFromStream be a -> IO (Maybe ByteString) -> ReaderT r m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer = m () -> ReaderT r m ()
forall (m :: * -> *) a. Monad m => m a -> ReaderT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall a. SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyFromStream be m =>
SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer)

instance (MonadBeamCopyFromStream be m) => MonadBeamCopyFromStream be (Lazy.StateT r m) where
  runCopyFromStream :: forall a.
SqlCopyFromStream be a -> IO (Maybe ByteString) -> StateT r m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer = m () -> StateT r m ()
forall (m :: * -> *) a. Monad m => m a -> StateT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall a. SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyFromStream be m =>
SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer)

instance (MonadBeamCopyFromStream be m) => MonadBeamCopyFromStream be (Strict.StateT r m) where
  runCopyFromStream :: forall a.
SqlCopyFromStream be a -> IO (Maybe ByteString) -> StateT r m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer = m () -> StateT r m ()
forall (m :: * -> *) a. Monad m => m a -> StateT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall a. SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyFromStream be m =>
SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer)

instance (MonadBeamCopyFromStream be m, Monoid r) => MonadBeamCopyFromStream be (Lazy.WriterT r m) where
  runCopyFromStream :: forall a.
SqlCopyFromStream be a -> IO (Maybe ByteString) -> WriterT r m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer = m () -> WriterT r m ()
forall (m :: * -> *) a. Monad m => m a -> WriterT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall a. SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyFromStream be m =>
SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer)

instance (MonadBeamCopyFromStream be m, Monoid r) => MonadBeamCopyFromStream be (Strict.WriterT r m) where
  runCopyFromStream :: forall a.
SqlCopyFromStream be a -> IO (Maybe ByteString) -> WriterT r m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer = m () -> WriterT r m ()
forall (m :: * -> *) a. Monad m => m a -> WriterT r m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall a. SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyFromStream be m =>
SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer)

instance (MonadBeamCopyFromStream be m, Monoid w) => MonadBeamCopyFromStream be (Lazy.RWST r w s m) where
  runCopyFromStream :: forall a.
SqlCopyFromStream be a -> IO (Maybe ByteString) -> RWST r w s m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer = m () -> RWST r w s m ()
forall (m :: * -> *) a. Monad m => m a -> RWST r w s m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall a. SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyFromStream be m =>
SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer)

instance (MonadBeamCopyFromStream be m, Monoid w) => MonadBeamCopyFromStream be (Strict.RWST r w s m) where
  runCopyFromStream :: forall a.
SqlCopyFromStream be a -> IO (Maybe ByteString) -> RWST r w s m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer = m () -> RWST r w s m ()
forall (m :: * -> *) a. Monad m => m a -> RWST r w s m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall a. SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
forall be (m :: * -> *) a.
MonadBeamCopyFromStream be m =>
SqlCopyFromStream be a -> IO (Maybe ByteString) -> m ()
runCopyFromStream SqlCopyFromStream be a
s IO (Maybe ByteString)
producer)