-- | Tests for the 'EventSource' and 'EventSink' implementation in 'Hydra.Events.SQLiteBased'.
module Hydra.Events.SQLiteBasedSpec where

import Hydra.Prelude hiding (label)
import Test.Hydra.Prelude

import Cardano.Binary (decodeFull')
import Data.Aeson qualified as Aeson
import Data.ByteString qualified as BS
import Data.List (zipWith3)
import Data.List qualified as List
import Database.SQLite.Simple (Only (..), close, execute, execute_, open, query)
import Hydra.Events (EventSink (..), EventSource (..), getEvents, putEvent)
import Hydra.Events.Rotation (EventStore (..))
import Hydra.Events.SQLiteBased (EventDecodingException, SQLiteLog (..), getSchemaVersion, nextVersion, withSQLiteEventStore)
import Hydra.HeadLogic.Outcome (StateChanged)
import Hydra.HeadLogic.StateEvent (StateEvent (..))
import Hydra.Ledger.Cardano (Tx)
import Hydra.Ledger.Simple (SimpleTx)
import Hydra.Logging (Envelope (..), nullTracer)
import System.Directory (doesFileExist, getFileSize)
import Test.Hydra.Chain.Direct.State ()
import Test.Hydra.HeadLogic.StateEvent ()
import Test.Hydra.Ledger.Simple ()
import Test.QuickCheck (forAllShrink, generate, ioProperty, sublistOf, suchThat, vectorOf, (===))
import Test.QuickCheck.Gen (listOf)
import Test.Util (captureTracer)

spec :: Spec
spec :: Spec
spec = do
  FilePath -> Spec -> Spec
forall a. HasCallStack => FilePath -> SpecWith a -> SpecWith a
describe FilePath
"mkSQLiteEventStore" (Spec -> Spec) -> Spec -> Spec
forall a b. (a -> b) -> a -> b
$ do
    FilePath -> Property -> Spec
forall prop.
(HasCallStack, Testable prop) =>
FilePath -> prop -> Spec
prop FilePath
"can stream events" (Property -> Spec) -> Property -> Spec
forall a b. (a -> b) -> a -> b
$
      Gen [StateEvent SimpleTx]
-> ([StateEvent SimpleTx] -> [[StateEvent SimpleTx]])
-> ([StateEvent SimpleTx] -> Property)
-> Property
forall a prop.
(Show a, Testable prop) =>
Gen a -> (a -> [a]) -> (a -> prop) -> Property
forAllShrink Gen [StateEvent SimpleTx]
genContinuousEvents [StateEvent SimpleTx] -> [[StateEvent SimpleTx]]
forall a. Arbitrary a => a -> [a]
shrink (([StateEvent SimpleTx] -> Property) -> Property)
-> ([StateEvent SimpleTx] -> Property) -> Property
forall a b. (a -> b) -> a -> b
$ \[StateEvent SimpleTx]
events ->
        IO Property -> Property
forall prop. Testable prop => IO prop -> Property
ioProperty (IO Property -> Property) -> IO Property -> Property
forall a b. (a -> b) -> a -> b
$
          (EventSource (StateEvent SimpleTx) IO
 -> EventSink (StateEvent SimpleTx) IO -> IO Property)
-> IO Property
forall b.
(EventSource (StateEvent SimpleTx) IO
 -> EventSink (StateEvent SimpleTx) IO -> IO b)
-> IO b
withEventSourceAndSink ((EventSource (StateEvent SimpleTx) IO
  -> EventSink (StateEvent SimpleTx) IO -> IO Property)
 -> IO Property)
-> (EventSource (StateEvent SimpleTx) IO
    -> EventSink (StateEvent SimpleTx) IO -> IO Property)
-> IO Property
forall a b. (a -> b) -> a -> b
$ \EventSource{HasEventId (StateEvent SimpleTx) =>
ConduitT () (StateEvent SimpleTx) (ResourceT IO) ()
sourceEvents :: HasEventId (StateEvent SimpleTx) =>
ConduitT () (StateEvent SimpleTx) (ResourceT IO) ()
$sel:sourceEvents:EventSource :: forall e (m :: * -> *).
EventSource e m -> HasEventId e => ConduitT () e (ResourceT m) ()
sourceEvents} EventSink{HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
$sel:putEvent:EventSink :: forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent :: HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
putEvent} -> do
            [StateEvent SimpleTx] -> (StateEvent SimpleTx -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [StateEvent SimpleTx]
events StateEvent SimpleTx -> IO ()
HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
putEvent
            -- XXX: Should assert while streaming
            [StateEvent SimpleTx]
streamedEvents <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents ((HasEventId (StateEvent SimpleTx) =>
 ConduitT () (StateEvent SimpleTx) (ResourceT IO) ())
-> EventSource (StateEvent SimpleTx) IO
forall e (m :: * -> *).
(HasEventId e => ConduitT () e (ResourceT m) ()) -> EventSource e m
EventSource ConduitT () (StateEvent SimpleTx) (ResourceT IO) ()
HasEventId (StateEvent SimpleTx) =>
ConduitT () (StateEvent SimpleTx) (ResourceT IO) ()
sourceEvents)
            Property -> IO Property
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Property -> IO Property) -> Property -> IO Property
forall a b. (a -> b) -> a -> b
$
              [StateEvent SimpleTx]
streamedEvents [StateEvent SimpleTx] -> [StateEvent SimpleTx] -> Property
forall a. (Eq a, Show a) => a -> a -> Property
=== [StateEvent SimpleTx]
events

    FilePath -> Property -> Spec
forall prop.
(HasCallStack, Testable prop) =>
FilePath -> prop -> Spec
prop FilePath
"can handle continuous events" (Property -> Spec) -> Property -> Spec
forall a b. (a -> b) -> a -> b
$
      Gen [StateEvent SimpleTx]
-> ([StateEvent SimpleTx] -> [[StateEvent SimpleTx]])
-> ([StateEvent SimpleTx] -> Property)
-> Property
forall a prop.
(Show a, Testable prop) =>
Gen a -> (a -> [a]) -> (a -> prop) -> Property
forAllShrink Gen [StateEvent SimpleTx]
genContinuousEvents [StateEvent SimpleTx] -> [[StateEvent SimpleTx]]
forall a. Arbitrary a => a -> [a]
shrink (([StateEvent SimpleTx] -> Property) -> Property)
-> ([StateEvent SimpleTx] -> Property) -> Property
forall a b. (a -> b) -> a -> b
$ \[StateEvent SimpleTx]
events ->
        IO Property -> Property
forall prop. Testable prop => IO prop -> Property
ioProperty (IO Property -> Property) -> IO Property -> Property
forall a b. (a -> b) -> a -> b
$ do
          (EventSource (StateEvent SimpleTx) IO
 -> EventSink (StateEvent SimpleTx) IO -> IO Property)
-> IO Property
forall b.
(EventSource (StateEvent SimpleTx) IO
 -> EventSink (StateEvent SimpleTx) IO -> IO b)
-> IO b
withEventSourceAndSink ((EventSource (StateEvent SimpleTx) IO
  -> EventSink (StateEvent SimpleTx) IO -> IO Property)
 -> IO Property)
-> (EventSource (StateEvent SimpleTx) IO
    -> EventSink (StateEvent SimpleTx) IO -> IO Property)
-> IO Property
forall a b. (a -> b) -> a -> b
$ \EventSource (StateEvent SimpleTx) IO
src EventSink{HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
$sel:putEvent:EventSink :: forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent :: HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
putEvent} -> do
            [StateEvent SimpleTx] -> (StateEvent SimpleTx -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [StateEvent SimpleTx]
events StateEvent SimpleTx -> IO ()
HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
putEvent
            [StateEvent SimpleTx]
loadedEvents <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource (StateEvent SimpleTx) IO
src
            Property -> IO Property
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Property -> IO Property) -> Property -> IO Property
forall a b. (a -> b) -> a -> b
$
              [StateEvent SimpleTx]
loadedEvents [StateEvent SimpleTx] -> [StateEvent SimpleTx] -> Property
forall a. (Eq a, Show a) => a -> a -> Property
=== [StateEvent SimpleTx]
events

    FilePath -> Property -> Spec
forall prop.
(HasCallStack, Testable prop) =>
FilePath -> prop -> Spec
prop FilePath
"can handle non-continuous events" (Property -> Spec) -> Property -> Spec
forall a b. (a -> b) -> a -> b
$
      Gen [StateEvent SimpleTx]
-> ([StateEvent SimpleTx] -> [[StateEvent SimpleTx]])
-> ([StateEvent SimpleTx] -> Property)
-> Property
forall a prop.
(Show a, Testable prop) =>
Gen a -> (a -> [a]) -> (a -> prop) -> Property
forAllShrink ([StateEvent SimpleTx] -> Gen [StateEvent SimpleTx]
forall a. [a] -> Gen [a]
sublistOf ([StateEvent SimpleTx] -> Gen [StateEvent SimpleTx])
-> Gen [StateEvent SimpleTx] -> Gen [StateEvent SimpleTx]
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< Gen [StateEvent SimpleTx]
genContinuousEvents) [StateEvent SimpleTx] -> [[StateEvent SimpleTx]]
forall a. Arbitrary a => a -> [a]
shrink (([StateEvent SimpleTx] -> Property) -> Property)
-> ([StateEvent SimpleTx] -> Property) -> Property
forall a b. (a -> b) -> a -> b
$ \[StateEvent SimpleTx]
events ->
        IO Property -> Property
forall prop. Testable prop => IO prop -> Property
ioProperty (IO Property -> Property) -> IO Property -> Property
forall a b. (a -> b) -> a -> b
$ do
          (EventSource (StateEvent SimpleTx) IO
 -> EventSink (StateEvent SimpleTx) IO -> IO Property)
-> IO Property
forall b.
(EventSource (StateEvent SimpleTx) IO
 -> EventSink (StateEvent SimpleTx) IO -> IO b)
-> IO b
withEventSourceAndSink ((EventSource (StateEvent SimpleTx) IO
  -> EventSink (StateEvent SimpleTx) IO -> IO Property)
 -> IO Property)
-> (EventSource (StateEvent SimpleTx) IO
    -> EventSink (StateEvent SimpleTx) IO -> IO Property)
-> IO Property
forall a b. (a -> b) -> a -> b
$ \EventSource (StateEvent SimpleTx) IO
src EventSink{HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
$sel:putEvent:EventSink :: forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent :: HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
putEvent} -> do
            [StateEvent SimpleTx] -> (StateEvent SimpleTx -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [StateEvent SimpleTx]
events StateEvent SimpleTx -> IO ()
HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
putEvent
            [StateEvent SimpleTx]
loadedEvents <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource (StateEvent SimpleTx) IO
src
            Property -> IO Property
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Property -> IO Property) -> Property -> IO Property
forall a b. (a -> b) -> a -> b
$
              [StateEvent SimpleTx]
loadedEvents [StateEvent SimpleTx] -> [StateEvent SimpleTx] -> Property
forall a. (Eq a, Show a) => a -> a -> Property
=== [StateEvent SimpleTx]
events

    FilePath -> Property -> Spec
forall prop.
(HasCallStack, Testable prop) =>
FilePath -> prop -> Spec
prop FilePath
"can handle duplicate events" (Property -> Spec) -> Property -> Spec
forall a b. (a -> b) -> a -> b
$
      Gen [StateEvent SimpleTx]
-> ([StateEvent SimpleTx] -> [[StateEvent SimpleTx]])
-> ([StateEvent SimpleTx] -> Property)
-> Property
forall a prop.
(Show a, Testable prop) =>
Gen a -> (a -> [a]) -> (a -> prop) -> Property
forAllShrink Gen [StateEvent SimpleTx]
genContinuousEvents [StateEvent SimpleTx] -> [[StateEvent SimpleTx]]
forall a. Arbitrary a => a -> [a]
shrink (([StateEvent SimpleTx] -> Property) -> Property)
-> ([StateEvent SimpleTx] -> Property) -> Property
forall a b. (a -> b) -> a -> b
$ \[StateEvent SimpleTx]
events ->
        IO Property -> Property
forall prop. Testable prop => IO prop -> Property
ioProperty (IO Property -> Property) -> IO Property -> Property
forall a b. (a -> b) -> a -> b
$
          (EventSource (StateEvent SimpleTx) IO
 -> EventSink (StateEvent SimpleTx) IO -> IO Property)
-> IO Property
forall b.
(EventSource (StateEvent SimpleTx) IO
 -> EventSink (StateEvent SimpleTx) IO -> IO b)
-> IO b
withEventSourceAndSink ((EventSource (StateEvent SimpleTx) IO
  -> EventSink (StateEvent SimpleTx) IO -> IO Property)
 -> IO Property)
-> (EventSource (StateEvent SimpleTx) IO
    -> EventSink (StateEvent SimpleTx) IO -> IO Property)
-> IO Property
forall a b. (a -> b) -> a -> b
$ \EventSource (StateEvent SimpleTx) IO
src EventSink{HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
$sel:putEvent:EventSink :: forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent :: HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
putEvent} -> do
            [StateEvent SimpleTx] -> (StateEvent SimpleTx -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [StateEvent SimpleTx]
events StateEvent SimpleTx -> IO ()
HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
putEvent
            [StateEvent SimpleTx]
loadedEvents <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource (StateEvent SimpleTx) IO
src
            -- Put the loaded events again (as the node would do)
            [StateEvent SimpleTx] -> (StateEvent SimpleTx -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [StateEvent SimpleTx]
loadedEvents StateEvent SimpleTx -> IO ()
HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
putEvent
            [StateEvent SimpleTx]
allEvents <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource (StateEvent SimpleTx) IO
src
            Property -> IO Property
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Property -> IO Property) -> Property -> IO Property
forall a b. (a -> b) -> a -> b
$
              [StateEvent SimpleTx]
allEvents [StateEvent SimpleTx] -> [StateEvent SimpleTx] -> Property
forall a. (Eq a, Show a) => a -> a -> Property
=== [StateEvent SimpleTx]
loadedEvents

    FilePath -> IO () -> SpecWith (Arg (IO ()))
forall a.
(HasCallStack, Example a) =>
FilePath -> a -> SpecWith (Arg a)
it FilePath
"throws EventDecodingException on invalid data in database" (IO () -> SpecWith (Arg (IO ())))
-> IO () -> SpecWith (Arg (IO ()))
forall a b. (a -> b) -> a -> b
$ do
      FilePath -> (FilePath -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
FilePath -> (FilePath -> m r) -> m r
withTempDir FilePath
"hydra-sqlite-persistence" ((FilePath -> IO ()) -> IO ()) -> (FilePath -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \FilePath
tmpDir -> do
        let dbFile :: FilePath
dbFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/hydra.db"
            stateFile :: FilePath
stateFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/state"
        forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore @(StateEvent SimpleTx) Tracer IO SQLiteLog
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer FilePath
dbFile FilePath
stateFile ((EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ())
-> (EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EventStore (StateEvent SimpleTx) IO
store -> do
          -- Insert a row with undecodable data directly via a separate connection
          IO Connection
-> (Connection -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (FilePath -> IO Connection
open FilePath
dbFile) Connection -> IO ()
close ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
            Connection -> Query -> (EventId, ByteString) -> IO ()
forall q. ToRow q => Connection -> Query -> q -> IO ()
execute Connection
conn Query
"INSERT INTO events (event_id, event_data) VALUES (?, ?)" (EventId
1 :: Word64, ByteString
"not valid cbor" :: ByteString)
          EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents (EventStore (StateEvent SimpleTx) IO
-> EventSource (StateEvent SimpleTx) IO
forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource EventStore (StateEvent SimpleTx) IO
store)
            IO [StateEvent SimpleTx]
-> Selector EventDecodingException -> IO ()
forall e a.
(HasCallStack, Exception e) =>
IO a -> Selector e -> IO ()
`shouldThrow` \(EventDecodingException
_ :: EventDecodingException) -> Bool
True

    FilePath -> IO () -> SpecWith (Arg (IO ()))
forall a.
(HasCallStack, Example a) =>
FilePath -> a -> SpecWith (Arg a)
it FilePath
"throws EventDecodingException on invalid lines during migration" (IO () -> SpecWith (Arg (IO ())))
-> IO () -> SpecWith (Arg (IO ()))
forall a b. (a -> b) -> a -> b
$ do
      FilePath -> (FilePath -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
FilePath -> (FilePath -> m r) -> m r
withTempDir FilePath
"hydra-sqlite-persistence" ((FilePath -> IO ()) -> IO ()) -> (FilePath -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \FilePath
tmpDir -> do
        let legacyFile :: FilePath
legacyFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/state"
        let dbFile :: FilePath
dbFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/hydra.db"
        FilePath -> ByteString -> IO ()
forall (m :: * -> *). MonadIO m => FilePath -> ByteString -> m ()
writeFileBS FilePath
legacyFile ByteString
"{invalid json\n"
        forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore @(StateEvent SimpleTx) Tracer IO SQLiteLog
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer FilePath
dbFile FilePath
legacyFile (\EventStore (StateEvent SimpleTx) IO
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
          IO () -> Selector EventDecodingException -> IO ()
forall e a.
(HasCallStack, Exception e) =>
IO a -> Selector e -> IO ()
`shouldThrow` \(EventDecodingException
_ :: EventDecodingException) -> Bool
True

    FilePath -> IO () -> SpecWith (Arg (IO ()))
forall a.
(HasCallStack, Example a) =>
FilePath -> a -> SpecWith (Arg a)
it FilePath
"fresh database ends up at nextVersion" (IO () -> SpecWith (Arg (IO ())))
-> IO () -> SpecWith (Arg (IO ()))
forall a b. (a -> b) -> a -> b
$ do
      FilePath -> (FilePath -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
FilePath -> (FilePath -> m r) -> m r
withTempDir FilePath
"hydra-sqlite-persistence" ((FilePath -> IO ()) -> IO ()) -> (FilePath -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \FilePath
tmpDir -> do
        let dbFile :: FilePath
dbFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/hydra.db"
            stateFile :: FilePath
stateFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/state"
        forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore @(StateEvent SimpleTx) Tracer IO SQLiteLog
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer FilePath
dbFile FilePath
stateFile ((EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ())
-> (EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EventStore (StateEvent SimpleTx) IO
_ -> do
          Int
v <- IO Connection
-> (Connection -> IO ()) -> (Connection -> IO Int) -> IO Int
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (FilePath -> IO Connection
open FilePath
dbFile) Connection -> IO ()
close Connection -> IO Int
getSchemaVersion
          Int
v Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Int
nextVersion

    FilePath -> IO () -> SpecWith (Arg (IO ()))
forall a.
(HasCallStack, Example a) =>
FilePath -> a -> SpecWith (Arg a)
it FilePath
"opening the database twice does not fail" (IO () -> SpecWith (Arg (IO ())))
-> IO () -> SpecWith (Arg (IO ()))
forall a b. (a -> b) -> a -> b
$ do
      FilePath -> (FilePath -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
FilePath -> (FilePath -> m r) -> m r
withTempDir FilePath
"hydra-sqlite-persistence" ((FilePath -> IO ()) -> IO ()) -> (FilePath -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \FilePath
tmpDir -> do
        let dbFile :: FilePath
dbFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/hydra.db"
            stateFile :: FilePath
stateFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/state"
        forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore @(StateEvent SimpleTx) Tracer IO SQLiteLog
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer FilePath
dbFile FilePath
stateFile ((EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ())
-> (EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EventStore (StateEvent SimpleTx) IO
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore @(StateEvent SimpleTx) Tracer IO SQLiteLog
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer FilePath
dbFile FilePath
stateFile ((EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ())
-> (EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EventStore (StateEvent SimpleTx) IO
_ -> do
          Int
v <- IO Connection
-> (Connection -> IO ()) -> (Connection -> IO Int) -> IO Int
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (FilePath -> IO Connection
open FilePath
dbFile) Connection -> IO ()
close Connection -> IO Int
getSchemaVersion
          Int
v Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Int
nextVersion

    FilePath -> IO () -> SpecWith (Arg (IO ()))
forall a.
(HasCallStack, Example a) =>
FilePath -> a -> SpecWith (Arg a)
it FilePath
"rejects a database with a newer schema version" (IO () -> SpecWith (Arg (IO ())))
-> IO () -> SpecWith (Arg (IO ()))
forall a b. (a -> b) -> a -> b
$ do
      FilePath -> (FilePath -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
FilePath -> (FilePath -> m r) -> m r
withTempDir FilePath
"hydra-sqlite-persistence" ((FilePath -> IO ()) -> IO ()) -> (FilePath -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \FilePath
tmpDir -> do
        let dbFile :: FilePath
dbFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/hydra.db"
            stateFile :: FilePath
stateFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/state"
        -- Create a DB with a version beyond what we know
        IO Connection
-> (Connection -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (FilePath -> IO Connection
open FilePath
dbFile) Connection -> IO ()
close ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
          Connection -> Query -> IO ()
execute_ Connection
conn (Query -> IO ()) -> Query -> IO ()
forall a b. (a -> b) -> a -> b
$ FilePath -> Query
forall a. IsString a => FilePath -> a
fromString (FilePath -> Query) -> FilePath -> Query
forall a b. (a -> b) -> a -> b
$ FilePath
"PRAGMA user_version = " FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> Int -> FilePath
forall b a. (Show a, IsString b) => a -> b
show (Int
nextVersion Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1)
        forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore @(StateEvent SimpleTx) Tracer IO SQLiteLog
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer FilePath
dbFile FilePath
stateFile (\EventStore (StateEvent SimpleTx) IO
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
          IO () -> Selector ErrorCall -> IO ()
forall e a.
(HasCallStack, Exception e) =>
IO a -> Selector e -> IO ()
`shouldThrow` Selector ErrorCall
anyErrorCall

    FilePath -> IO () -> SpecWith (Arg (IO ()))
forall a.
(HasCallStack, Example a) =>
FilePath -> a -> SpecWith (Arg a)
it FilePath
"migrates a v1 JSON database to v2 CBOR" (IO () -> SpecWith (Arg (IO ())))
-> IO () -> SpecWith (Arg (IO ()))
forall a b. (a -> b) -> a -> b
$
      Int -> IO ()
migratesV1JsonToV2CBOR Int
50

    FilePath -> IO () -> SpecWith (Arg (IO ()))
forall a.
(HasCallStack, Example a) =>
FilePath -> a -> SpecWith (Arg a)
it FilePath
"aborts migration and keeps v1 intact on a corrupt row" (IO () -> SpecWith (Arg (IO ())))
-> IO () -> SpecWith (Arg (IO ()))
forall a b. (a -> b) -> a -> b
$ do
      FilePath -> (FilePath -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
FilePath -> (FilePath -> m r) -> m r
withTempDir FilePath
"hydra-sqlite-persistence" ((FilePath -> IO ()) -> IO ()) -> (FilePath -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \FilePath
tmpDir -> do
        let dbFile :: FilePath
dbFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/hydra.db"
            stateFile :: FilePath
stateFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/state"
        StateEvent SimpleTx
goodEvent :: StateEvent SimpleTx <- Gen (StateEvent SimpleTx) -> IO (StateEvent SimpleTx)
forall a. Gen a -> IO a
generate Gen (StateEvent SimpleTx)
forall a. Arbitrary a => Gen a
arbitrary
        IO Connection
-> (Connection -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (FilePath -> IO Connection
open FilePath
dbFile) Connection -> IO ()
close ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn -> do
          Connection -> Query -> IO ()
execute_ Connection
conn Query
"CREATE TABLE events (event_id INTEGER NOT NULL PRIMARY KEY, event_data BLOB NOT NULL)"
          Connection -> Query -> IO ()
execute_ Connection
conn Query
"PRAGMA user_version = 1"
          Connection -> Query -> (EventId, ByteString) -> IO ()
forall q. ToRow q => Connection -> Query -> q -> IO ()
execute Connection
conn Query
"INSERT INTO events (event_id, event_data) VALUES (?, ?)" (EventId
1 :: Word64, ByteString -> ByteString
forall l s. LazyStrict l s => l -> s
toStrict (StateEvent SimpleTx -> ByteString
forall a. ToJSON a => a -> ByteString
Aeson.encode StateEvent SimpleTx
goodEvent))
          Connection -> Query -> (EventId, ByteString) -> IO ()
forall q. ToRow q => Connection -> Query -> q -> IO ()
execute Connection
conn Query
"INSERT INTO events (event_id, event_data) VALUES (?, ?)" (EventId
2 :: Word64, ByteString
"not valid json" :: ByteString)
        forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore @(StateEvent SimpleTx) Tracer IO SQLiteLog
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer FilePath
dbFile FilePath
stateFile (\EventStore (StateEvent SimpleTx) IO
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
          IO () -> Selector EventDecodingException -> IO ()
forall e a.
(HasCallStack, Exception e) =>
IO a -> Selector e -> IO ()
`shouldThrow` \(EventDecodingException
_ :: EventDecodingException) -> Bool
True
        -- The failed migration must roll back: still version 1, rows untouched.
        Int
v <- IO Connection
-> (Connection -> IO ()) -> (Connection -> IO Int) -> IO Int
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (FilePath -> IO Connection
open FilePath
dbFile) Connection -> IO ()
close Connection -> IO Int
getSchemaVersion
        Int
v Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Int
1

    FilePath -> Property -> Spec
forall prop.
(HasCallStack, Testable prop) =>
FilePath -> prop -> Spec
prop FilePath
"can migrate from file-based store" (Property -> Spec) -> Property -> Spec
forall a b. (a -> b) -> a -> b
$
      Gen [StateEvent SimpleTx]
-> ([StateEvent SimpleTx] -> [[StateEvent SimpleTx]])
-> ([StateEvent SimpleTx] -> Property)
-> Property
forall a prop.
(Show a, Testable prop) =>
Gen a -> (a -> [a]) -> (a -> prop) -> Property
forAllShrink Gen [StateEvent SimpleTx]
genContinuousEvents [StateEvent SimpleTx] -> [[StateEvent SimpleTx]]
forall a. Arbitrary a => a -> [a]
shrink (([StateEvent SimpleTx] -> Property) -> Property)
-> ([StateEvent SimpleTx] -> Property) -> Property
forall a b. (a -> b) -> a -> b
$ \[StateEvent SimpleTx]
events ->
        IO Property -> Property
forall prop. Testable prop => IO prop -> Property
ioProperty (IO Property -> Property) -> IO Property -> Property
forall a b. (a -> b) -> a -> b
$ do
          FilePath -> (FilePath -> IO Property) -> IO Property
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
FilePath -> (FilePath -> m r) -> m r
withTempDir FilePath
"hydra-sqlite-persistence" ((FilePath -> IO Property) -> IO Property)
-> (FilePath -> IO Property) -> IO Property
forall a b. (a -> b) -> a -> b
$ \FilePath
tmpDir -> do
            let legacyFile :: FilePath
legacyFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/state"
            let dbFile :: FilePath
dbFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/hydra.db"
            -- Write events to the legacy file
            [StateEvent SimpleTx] -> (StateEvent SimpleTx -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [StateEvent SimpleTx]
events ((StateEvent SimpleTx -> IO ()) -> IO ())
-> (StateEvent SimpleTx -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \StateEvent SimpleTx
e ->
              FilePath -> ByteString -> IO ()
BS.appendFile FilePath
legacyFile (ByteString -> ByteString
forall l s. LazyStrict l s => l -> s
toStrict (StateEvent SimpleTx -> ByteString
forall a. ToJSON a => a -> ByteString
Aeson.encode StateEvent SimpleTx
e) ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> ByteString
"\n")
            -- Migrate into SQLite
            (Tracer IO SQLiteLog
tracer, IO [Envelope SQLiteLog]
getTraces) <- Text -> IO (Tracer IO SQLiteLog, IO [Envelope SQLiteLog])
forall a. Text -> IO (Tracer IO a, IO [Envelope a])
captureTracer Text
"sqlite"
            Tracer IO SQLiteLog
-> FilePath
-> FilePath
-> (EventStore (StateEvent SimpleTx) IO -> IO Property)
-> IO Property
forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore Tracer IO SQLiteLog
tracer FilePath
dbFile FilePath
legacyFile ((EventStore (StateEvent SimpleTx) IO -> IO Property)
 -> IO Property)
-> (EventStore (StateEvent SimpleTx) IO -> IO Property)
-> IO Property
forall a b. (a -> b) -> a -> b
$ \EventStore (StateEvent SimpleTx) IO
store -> do
              -- Verify all events are present
              [StateEvent SimpleTx]
loadedEvents <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents (EventStore (StateEvent SimpleTx) IO
-> EventSource (StateEvent SimpleTx) IO
forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource EventStore (StateEvent SimpleTx) IO
store)
              -- Verify migration was logged
              [Envelope SQLiteLog]
traces <- IO [Envelope SQLiteLog]
getTraces
              let msgs :: [SQLiteLog]
msgs = (Envelope SQLiteLog -> SQLiteLog)
-> [Envelope SQLiteLog] -> [SQLiteLog]
forall a b. (a -> b) -> [a] -> [b]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap Envelope SQLiteLog -> SQLiteLog
forall a. Envelope a -> a
message [Envelope SQLiteLog]
traces
              Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([StateEvent SimpleTx] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [StateEvent SimpleTx]
events) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
                [SQLiteLog]
msgs [SQLiteLog] -> ([SQLiteLog] -> Bool) -> IO ()
forall a. (HasCallStack, Show a) => a -> (a -> Bool) -> IO ()
`shouldSatisfy` SQLiteLog -> [SQLiteLog] -> Bool
forall (f :: * -> *) a.
(Foldable f, DisallowElem f, Eq a) =>
a -> f a -> Bool
elem MigrationComplete{FilePath
legacyFile :: FilePath
$sel:legacyFile:MigratingFromFileBased :: FilePath
legacyFile}
              Property -> IO Property
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Property -> IO Property) -> Property -> IO Property
forall a b. (a -> b) -> a -> b
$
                [StateEvent SimpleTx]
loadedEvents [StateEvent SimpleTx] -> [StateEvent SimpleTx] -> Property
forall a. (Eq a, Show a) => a -> a -> Property
=== [StateEvent SimpleTx]
events

    FilePath -> IO () -> SpecWith (Arg (IO ()))
forall a.
(HasCallStack, Example a) =>
FilePath -> a -> SpecWith (Arg a)
it FilePath
"keeps a backup of the database before rotating" (IO () -> SpecWith (Arg (IO ())))
-> IO () -> SpecWith (Arg (IO ()))
forall a b. (a -> b) -> a -> b
$ do
      FilePath -> (FilePath -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
FilePath -> (FilePath -> m r) -> m r
withTempDir FilePath
"hydra-sqlite-persistence" ((FilePath -> IO ()) -> IO ()) -> (FilePath -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \FilePath
tmpDir -> do
        let dbFile :: FilePath
dbFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/hydra.db"
            stateFile :: FilePath
stateFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/state"
        [StateEvent SimpleTx]
events <- Gen [StateEvent SimpleTx] -> IO [StateEvent SimpleTx]
forall a. Gen a -> IO a
generate (Gen [StateEvent SimpleTx] -> IO [StateEvent SimpleTx])
-> Gen [StateEvent SimpleTx] -> IO [StateEvent SimpleTx]
forall a b. (a -> b) -> a -> b
$ Gen [StateEvent SimpleTx]
genContinuousEvents Gen [StateEvent SimpleTx]
-> ([StateEvent SimpleTx] -> Bool) -> Gen [StateEvent SimpleTx]
forall a. Gen a -> (a -> Bool) -> Gen a
`suchThat` (Bool -> Bool
not (Bool -> Bool)
-> ([StateEvent SimpleTx] -> Bool) -> [StateEvent SimpleTx] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [StateEvent SimpleTx] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null)
        let lastId :: EventId
lastId = StateEvent SimpleTx -> EventId
forall tx. StateEvent tx -> EventId
eventId ([StateEvent SimpleTx] -> StateEvent SimpleTx
forall a. HasCallStack => [a] -> a
List.last [StateEvent SimpleTx]
events)
            -- content is irrelevant here; we only assert on the archived events
            checkpointEvent :: StateEvent SimpleTx
checkpointEvent = [StateEvent SimpleTx] -> StateEvent SimpleTx
forall a. HasCallStack => [a] -> a
List.last [StateEvent SimpleTx]
events
        forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore @(StateEvent SimpleTx) Tracer IO SQLiteLog
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer FilePath
dbFile FilePath
stateFile ((EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ())
-> (EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EventStore{$sel:eventSink:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSink e m
eventSink = EventSink{HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
$sel:putEvent:EventSink :: forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent :: HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
putEvent}, EventSource (StateEvent SimpleTx) IO
$sel:eventSource:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource :: EventSource (StateEvent SimpleTx) IO
eventSource, EventId -> StateEvent SimpleTx -> IO ()
rotate :: EventId -> StateEvent SimpleTx -> IO ()
$sel:rotate:EventStore :: forall e (m :: * -> *). EventStore e m -> EventId -> e -> m ()
rotate} -> do
          [StateEvent SimpleTx] -> (StateEvent SimpleTx -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [StateEvent SimpleTx]
events StateEvent SimpleTx -> IO ()
HasEventId (StateEvent SimpleTx) => StateEvent SimpleTx -> IO ()
putEvent
          EventId -> StateEvent SimpleTx -> IO ()
rotate EventId
lastId StateEvent SimpleTx
checkpointEvent
          -- the active store holds only the checkpoint after rotation
          [StateEvent SimpleTx]
active <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource (StateEvent SimpleTx) IO
eventSource
          [StateEvent SimpleTx]
active [StateEvent SimpleTx] -> [StateEvent SimpleTx] -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` [StateEvent SimpleTx
checkpointEvent]
        -- the pre-rotation events are retained in the numbered backup
        let backupPath :: FilePath
backupPath = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/old-state/hydra-" FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> EventId -> FilePath
forall b a. (Show a, IsString b) => a -> b
show EventId
lastId FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
".db"
        FilePath -> IO Bool
doesFileExist FilePath
backupPath IO Bool -> Bool -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Bool
True
        forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore @(StateEvent SimpleTx) Tracer IO SQLiteLog
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer FilePath
backupPath FilePath
stateFile ((EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ())
-> (EventStore (StateEvent SimpleTx) IO -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EventStore{EventSource (StateEvent SimpleTx) IO
$sel:eventSource:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource :: EventSource (StateEvent SimpleTx) IO
eventSource} -> do
          [StateEvent SimpleTx]
backedUp <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource (StateEvent SimpleTx) IO
eventSource
          [StateEvent SimpleTx]
backedUp [StateEvent SimpleTx] -> [StateEvent SimpleTx] -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` [StateEvent SimpleTx]
events

-- | Create a v1 database with @n@ JSON-encoded 'StateEvent Tx' rows (as
-- written by hydra-node versions before the CBOR switch), open it (triggering
-- the migration) and assert nothing is lost: the events load back equal, and
-- each migrated row's raw CBOR decodes to exactly the event that was inserted
-- as JSON. Uses the production event type ('Tx', the one real hydra.db files
-- store) rather than 'SimpleTx'.
migratesV1JsonToV2CBOR :: Int -> IO ()
migratesV1JsonToV2CBOR :: Int -> IO ()
migratesV1JsonToV2CBOR Int
n =
  FilePath -> (FilePath -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
FilePath -> (FilePath -> m r) -> m r
withTempDir FilePath
"hydra-sqlite-persistence" ((FilePath -> IO ()) -> IO ()) -> (FilePath -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \FilePath
tmpDir -> do
    let dbFile :: FilePath
dbFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/hydra.db"
        stateFile :: FilePath
stateFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/state"
    [StateEvent Tx]
events <- Gen [StateEvent Tx] -> IO [StateEvent Tx]
forall a. Gen a -> IO a
generate (Gen [StateEvent Tx] -> IO [StateEvent Tx])
-> Gen [StateEvent Tx] -> IO [StateEvent Tx]
forall a b. (a -> b) -> a -> b
$ forall tx. [StateChanged tx] -> [UTCTime] -> [StateEvent tx]
mkContinuousEvents @Tx ([StateChanged Tx] -> [UTCTime] -> [StateEvent Tx])
-> Gen [StateChanged Tx] -> Gen ([UTCTime] -> [StateEvent Tx])
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Int -> Gen (StateChanged Tx) -> Gen [StateChanged Tx]
forall a. Int -> Gen a -> Gen [a]
vectorOf Int
n Gen (StateChanged Tx)
forall a. Arbitrary a => Gen a
arbitrary Gen ([UTCTime] -> [StateEvent Tx])
-> Gen [UTCTime] -> Gen [StateEvent Tx]
forall a b. Gen (a -> b) -> Gen a -> Gen b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Int -> Gen UTCTime -> Gen [UTCTime]
forall a. Int -> Gen a -> Gen [a]
vectorOf Int
n Gen UTCTime
forall a. Arbitrary a => Gen a
arbitrary
    IO Connection
-> (Connection -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (FilePath -> IO Connection
open FilePath
dbFile) Connection -> IO ()
close ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn -> do
      Connection -> Query -> IO ()
execute_ Connection
conn Query
"CREATE TABLE events (event_id INTEGER NOT NULL PRIMARY KEY, event_data BLOB NOT NULL)"
      Connection -> Query -> IO ()
execute_ Connection
conn Query
"PRAGMA user_version = 1"
      [StateEvent Tx] -> (StateEvent Tx -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [StateEvent Tx]
events ((StateEvent Tx -> IO ()) -> IO ())
-> (StateEvent Tx -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \StateEvent Tx
e ->
        Connection -> Query -> (EventId, ByteString) -> IO ()
forall q. ToRow q => Connection -> Query -> q -> IO ()
execute Connection
conn Query
"INSERT INTO events (event_id, event_data) VALUES (?, ?)" (StateEvent Tx -> EventId
forall tx. StateEvent tx -> EventId
eventId StateEvent Tx
e, ByteString -> ByteString
forall l s. LazyStrict l s => l -> s
toStrict (StateEvent Tx -> ByteString
forall a. ToJSON a => a -> ByteString
Aeson.encode StateEvent Tx
e))
    Integer
sizeBefore <- FilePath -> IO Integer
getFileSize FilePath
dbFile
    forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore @(StateEvent Tx) Tracer IO SQLiteLog
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer FilePath
dbFile FilePath
stateFile ((EventStore (StateEvent Tx) IO -> IO ()) -> IO ())
-> (EventStore (StateEvent Tx) IO -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EventStore (StateEvent Tx) IO
store -> do
      [StateEvent Tx]
loadedEvents <- EventSource (StateEvent Tx) IO -> IO [StateEvent Tx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents (EventStore (StateEvent Tx) IO -> EventSource (StateEvent Tx) IO
forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource EventStore (StateEvent Tx) IO
store)
      [StateEvent Tx]
loadedEvents [StateEvent Tx] -> [StateEvent Tx] -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` [StateEvent Tx]
events
    Int
v <- IO Connection
-> (Connection -> IO ()) -> (Connection -> IO Int) -> IO Int
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (FilePath -> IO Connection
open FilePath
dbFile) Connection -> IO ()
close Connection -> IO Int
getSchemaVersion
    Int
v Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Int
nextVersion
    -- Random events from the original JSON database must be found in the
    -- migrated database under the same event_id, with the row blob
    -- decoding (as CBOR) to exactly the event that was inserted.
    [StateEvent Tx]
picked <- Gen [StateEvent Tx] -> IO [StateEvent Tx]
forall a. Gen a -> IO a
generate (Gen [StateEvent Tx] -> IO [StateEvent Tx])
-> Gen [StateEvent Tx] -> IO [StateEvent Tx]
forall a b. (a -> b) -> a -> b
$ [StateEvent Tx] -> Gen [StateEvent Tx]
forall a. [a] -> Gen [a]
sublistOf [StateEvent Tx]
events Gen [StateEvent Tx]
-> ([StateEvent Tx] -> Bool) -> Gen [StateEvent Tx]
forall a. Gen a -> (a -> Bool) -> Gen a
`suchThat` (Bool -> Bool
not (Bool -> Bool)
-> ([StateEvent Tx] -> Bool) -> [StateEvent Tx] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [StateEvent Tx] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null)
    IO Connection
-> (Connection -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (FilePath -> IO Connection
open FilePath
dbFile) Connection -> IO ()
close ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
      [StateEvent Tx] -> (StateEvent Tx -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [StateEvent Tx]
picked ((StateEvent Tx -> IO ()) -> IO ())
-> (StateEvent Tx -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \StateEvent Tx
e -> do
        [Only ByteString]
rows :: [Only ByteString] <-
          Connection -> Query -> Only EventId -> IO [Only ByteString]
forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query Connection
conn Query
"SELECT event_data FROM events WHERE event_id = ?" (EventId -> Only EventId
forall a. a -> Only a
Only (StateEvent Tx -> EventId
forall tx. StateEvent tx -> EventId
eventId StateEvent Tx
e))
        case [Only ByteString]
rows of
          [Only ByteString
bytes] ->
            case ByteString -> Either DecoderError (StateEvent Tx)
forall a. FromCBOR a => ByteString -> Either DecoderError a
decodeFull' ByteString
bytes of
              Left DecoderError
err ->
                HasCallStack => FilePath -> IO ()
FilePath -> IO ()
expectationFailure (FilePath -> IO ()) -> FilePath -> IO ()
forall a b. (a -> b) -> a -> b
$
                  FilePath
"failed to decode migrated row " FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> EventId -> FilePath
forall b a. (Show a, IsString b) => a -> b
show (StateEvent Tx -> EventId
forall tx. StateEvent tx -> EventId
eventId StateEvent Tx
e) FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
" as CBOR: " FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> DecoderError -> FilePath
forall b a. (Show a, IsString b) => a -> b
show DecoderError
err
              Right (StateEvent Tx
decoded :: StateEvent Tx) -> StateEvent Tx
decoded StateEvent Tx -> StateEvent Tx -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` StateEvent Tx
e
          [Only ByteString]
_ ->
            HasCallStack => FilePath -> IO ()
FilePath -> IO ()
expectationFailure (FilePath -> IO ()) -> FilePath -> IO ()
forall a b. (a -> b) -> a -> b
$
              FilePath
"expected exactly one row for event_id " FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> EventId -> FilePath
forall b a. (Show a, IsString b) => a -> b
show (StateEvent Tx -> EventId
forall tx. StateEvent tx -> EventId
eventId StateEvent Tx
e) FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
", got " FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> Int -> FilePath
forall b a. (Show a, IsString b) => a -> b
show ([Only ByteString] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [Only ByteString]
rows)
    -- On real 'Tx' events transaction/UTxO bytes dominate and JSON stores
    -- them hex-encoded, so re-encoding to CBOR + VACUUM must strictly shrink
    -- the database — this locks in the compression win of the migration.
    Integer
sizeAfter <- FilePath -> IO Integer
getFileSize FilePath
dbFile
    Integer
sizeAfter Integer -> (Integer -> Bool) -> IO ()
forall a. (HasCallStack, Show a) => a -> (a -> Bool) -> IO ()
`shouldSatisfy` (Integer -> Integer -> Bool
forall a. Ord a => a -> a -> Bool
< Integer
sizeBefore)

genContinuousEvents :: Gen [StateEvent SimpleTx]
genContinuousEvents :: Gen [StateEvent SimpleTx]
genContinuousEvents =
  [StateChanged SimpleTx] -> [UTCTime] -> [StateEvent SimpleTx]
forall tx. [StateChanged tx] -> [UTCTime] -> [StateEvent tx]
mkContinuousEvents ([StateChanged SimpleTx] -> [UTCTime] -> [StateEvent SimpleTx])
-> Gen [StateChanged SimpleTx]
-> Gen ([UTCTime] -> [StateEvent SimpleTx])
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Gen (StateChanged SimpleTx) -> Gen [StateChanged SimpleTx]
forall a. Gen a -> Gen [a]
listOf Gen (StateChanged SimpleTx)
forall a. Arbitrary a => Gen a
arbitrary Gen ([UTCTime] -> [StateEvent SimpleTx])
-> Gen [UTCTime] -> Gen [StateEvent SimpleTx]
forall a b. Gen (a -> b) -> Gen a -> Gen b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Gen UTCTime -> Gen [UTCTime]
forall a. Gen a -> Gen [a]
listOf Gen UTCTime
forall a. Arbitrary a => Gen a
arbitrary

mkContinuousEvents :: [StateChanged tx] -> [UTCTime] -> [StateEvent tx]
mkContinuousEvents :: forall tx. [StateChanged tx] -> [UTCTime] -> [StateEvent tx]
mkContinuousEvents = (EventId -> StateChanged tx -> UTCTime -> StateEvent tx)
-> [EventId] -> [StateChanged tx] -> [UTCTime] -> [StateEvent tx]
forall a b c d. (a -> b -> c -> d) -> [a] -> [b] -> [c] -> [d]
zipWith3 EventId -> StateChanged tx -> UTCTime -> StateEvent tx
forall tx. EventId -> StateChanged tx -> UTCTime -> StateEvent tx
StateEvent [EventId
0 ..]

withEventSourceAndSink :: (EventSource (StateEvent SimpleTx) IO -> EventSink (StateEvent SimpleTx) IO -> IO b) -> IO b
withEventSourceAndSink :: forall b.
(EventSource (StateEvent SimpleTx) IO
 -> EventSink (StateEvent SimpleTx) IO -> IO b)
-> IO b
withEventSourceAndSink EventSource (StateEvent SimpleTx) IO
-> EventSink (StateEvent SimpleTx) IO -> IO b
action =
  FilePath -> (FilePath -> IO b) -> IO b
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
FilePath -> (FilePath -> m r) -> m r
withTempDir FilePath
"hydra-sqlite-persistence" ((FilePath -> IO b) -> IO b) -> (FilePath -> IO b) -> IO b
forall a b. (a -> b) -> a -> b
$ \FilePath
tmpDir -> do
    let dbFile :: FilePath
dbFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/hydra.db"
        stateFile :: FilePath
stateFile = FilePath
tmpDir FilePath -> FilePath -> FilePath
forall a. Semigroup a => a -> a -> a
<> FilePath
"/state"
    Tracer IO SQLiteLog
-> FilePath
-> FilePath
-> (EventStore (StateEvent SimpleTx) IO -> IO b)
-> IO b
forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore Tracer IO SQLiteLog
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer FilePath
dbFile FilePath
stateFile ((EventStore (StateEvent SimpleTx) IO -> IO b) -> IO b)
-> (EventStore (StateEvent SimpleTx) IO -> IO b) -> IO b
forall a b. (a -> b) -> a -> b
$ \EventStore{EventSource (StateEvent SimpleTx) IO
$sel:eventSource:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource :: EventSource (StateEvent SimpleTx) IO
eventSource, EventSink (StateEvent SimpleTx) IO
$sel:eventSink:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSink e m
eventSink :: EventSink (StateEvent SimpleTx) IO
eventSink} ->
      EventSource (StateEvent SimpleTx) IO
-> EventSink (StateEvent SimpleTx) IO -> IO b
action EventSource (StateEvent SimpleTx) IO
eventSource EventSink (StateEvent SimpleTx) IO
eventSink