wip

hbs2-git/hbs2-git-oracle/app/Main.hs

@@ -1,6 +1,7 @@
 module Main where
 
 import HBS2.Git.Oracle.Prelude
+import HBS2.Git.Oracle.Facts
 import HBS2.Git.Oracle.App
 import HBS2.Git.Oracle.Run
 

hbs2-git/hbs2-git-oracle/lib/HBS2/Git/Oracle/App.hs

@@ -5,14 +5,22 @@ module HBS2.Git.Oracle.App
   ( OracleEnv(..)
   , Oracle(..)
   , runWithOracleEnv
+  , withState
   ) where
 
 import HBS2.Git.Oracle.Prelude
+import HBS2.Git.Oracle.State
 
 import HBS2.Peer.CLI.Detect
 
-import GHC.TypeLits
+import HBS2.System.Dir
-import Codec.Serialise
+
+import DBPipe.SQLite
+
+import System.Directory
+
+myself :: FilePath
+myself = "hbs2-git-oracle"
 
 data OracleEnv =
   OracleEnv
@@ -23,6 +31,7 @@ data OracleEnv =
   , _refchanAPI    :: ServiceCaller RefChanAPI UNIX
   , _lwwAPI        :: ServiceCaller LWWRefAPI UNIX
   , _storage       :: AnyStorage
+  , _db            :: DBPipeEnv
   }
   deriving stock (Generic)
 
@@ -57,13 +66,25 @@ runWithOracleEnv rchan m = do
   storageAPI <- makeServiceCaller @StorageAPI (fromString soname)
   let sto = AnyStorage (StorageClient storageAPI)
 
+  let dbname = show (pretty (AsBase58 rchan))
+
+  dbpath <- liftIO (getXdgDirectory XdgData  myself)
+
+  let dbfile = dbpath </> dbname <> ".db"
+
+  mkdir dbpath
+
+  debug $ red "DBPATH" <+> pretty dbfile
+
+  db <- newDBPipeEnv dbPipeOptsDef dbfile
+
   env <- pure $ OracleEnv rchan
-                          -- author
                           peerAPI
                           reflogAPI
                           refchanAPI
                           lwwAPI
                           sto
+                          db
 
   let endpoints = [ Endpoint @UNIX  peerAPI
                   , Endpoint @UNIX  reflogAPI
@@ -78,5 +99,19 @@ runWithOracleEnv rchan m = do
 
     void $ ContT $ withAsync $ liftIO $ runReaderT (runServiceClientMulti endpoints) client
 
-    lift $ runReaderT (fromOracle m) env
+    lift $ runReaderT (fromOracle (withState evolveDB >> m)) env
+
+class Monad m => HasDB m where
+  getDB :: m DBPipeEnv
+
+instance Monad m => HasDB (Oracle m) where
+  getDB = asks _db
+
+
+withState :: forall m a . (MonadUnliftIO m, HasDB m)
+          => DBPipeM m a
+          -> m a
+withState dbAction = do
+  db <- getDB
+  withDB db dbAction
 

hbs2-git/hbs2-git-oracle/lib/HBS2/Git/Oracle/Facts.hs

@@ -0,0 +1,65 @@
+module HBS2.Git.Oracle.Facts where
+
+import HBS2.Git.Oracle.Prelude
+import Data.Word
+import Codec.Serialise
+
+type PKS = PubKey 'Sign HBS2Basic
+
+deriving instance Data (RefLogKey HBS2Basic)
+deriving instance Data (LWWRefKey HBS2Basic)
+
+data GitRepoRefFact =
+  GitRepoFact1
+  { gitLwwRef :: LWWRefKey HBS2Basic
+  , gitLwwSeq :: Word64
+  , gitRefLog :: RefLogKey HBS2Basic
+  }
+  deriving stock (Generic,Data)
+
+data GitRepoHeadFact =
+  GitRepoHeadFact1
+  { gitRepoHeadRef   :: HashRef
+  , gitRepoName      :: Text
+  , gitRepoBrief     :: Text
+  , gitRepoEncrypted :: Bool
+  }
+  deriving stock (Generic,Data)
+
+
+data GitRepoHeadVersionFact =
+  GitRepoHeadVersionFact1
+  { gitRepoHeadVersion  :: Word64
+  }
+  deriving stock (Generic,Data)
+
+data GitRepoFacts =
+      GitRepoRefFact  GitRepoRefFact
+    | GitRepoHeadFact HashRef GitRepoHeadFact
+    | GitRepoHeadVersionFact HashRef GitRepoHeadVersionFact
+    | GitRepoTxFact (LWWRefKey HBS2Basic) HashRef
+    deriving stock (Generic,Data)
+
+
+instance Serialise GitRepoRefFact
+instance Serialise GitRepoHeadFact
+instance Serialise GitRepoFacts
+instance Serialise GitRepoHeadVersionFact
+
+instance Pretty GitRepoFacts  where
+  pretty (GitRepoRefFact x)    = pretty x
+  pretty (GitRepoHeadFact ha x) = pretty ("gitrpoheadfact",ha,x)
+  pretty (GitRepoHeadVersionFact ha x) = pretty ("gitrpoheadversionfact",ha,x)
+  pretty (GitRepoTxFact r tx) = pretty ("gitrepotxfact", r, tx)
+
+instance Pretty GitRepoRefFact where
+  pretty (GitRepoFact1{..}) =
+    parens ( "gitrepofact1" <+>  hsep [pretty gitLwwRef, pretty gitLwwSeq, pretty gitRefLog])
+
+instance Pretty GitRepoHeadFact where
+  pretty (GitRepoHeadFact1{..}) =
+    parens ( "gitrepoheadfact1" <+>  hsep [pretty gitRepoHeadRef])
+
+instance Pretty GitRepoHeadVersionFact where
+  pretty (GitRepoHeadVersionFact1 v) = pretty v
+

hbs2-git/hbs2-git-oracle/lib/HBS2/Git/Oracle/Prelude.hs

@@ -25,6 +25,8 @@ module HBS2.Git.Oracle.Prelude
   , module HBS2.Peer.RPC.Client.StorageClient
   , module HBS2.Peer.RPC.Client.Unix
 
+  , module DBPipe.SQLite
+
   , module Data.Kind
   , module Control.Monad.Reader
   , module Control.Monad.Trans.Cont
@@ -57,9 +59,11 @@ import HBS2.Peer.RPC.API.Storage
 import HBS2.Peer.RPC.Client.StorageClient
 import HBS2.Peer.RPC.Client.Unix
 
+import DBPipe.SQLite hiding (runPipe)
+
 import Data.Kind
 import Control.Monad.Reader
-import Control.Monad.Trans.Cont
+import Control.Monad.Trans.Cont hiding (reset)
 import UnliftIO
 
 

hbs2-git/hbs2-git-oracle/lib/HBS2/Git/Oracle/Run.hs

@@ -5,6 +5,8 @@ module HBS2.Git.Oracle.Run where
 
 import HBS2.Git.Oracle.Prelude
 import HBS2.Git.Oracle.App
+import HBS2.Git.Oracle.Facts
+import HBS2.Git.Oracle.State
 
 import HBS2.Actors.Peer
 
@@ -25,7 +27,6 @@ import Lens.Micro.Platform hiding ( (.=) )
 
 import Data.Aeson as Aeson
 import Data.Aeson.Encode.Pretty qualified as A
-import Data.Word
 import Streaming.Prelude qualified as S
 import Codec.Serialise
 import Control.Monad.Trans.Maybe
@@ -37,56 +38,9 @@ import Data.ByteString.Lazy qualified as LBS
 import System.Process.Typed
 
 import System.Environment (getProgName, getArgs)
-import System.Exit
-
-type PKS = PubKey 'Sign HBS2Basic
 
 {- HLINT ignore "Functor law" -}
 
-deriving instance Data (RefLogKey HBS2Basic)
-deriving instance Data (LWWRefKey HBS2Basic)
-
-data GitRepoRefFact =
-  GitRepoFact1
-  { gitLwwRef :: LWWRefKey HBS2Basic
-  , gitLwwSeq :: Word64
-  , gitRefLog :: RefLogKey HBS2Basic
-  }
-  deriving stock (Generic,Data)
-
-data GitRepoHeadFact =
-  GitRepoHeadFact1
-  { gitRepoHeadRef   :: HashRef
-  , gitRepoName      :: Text
-  , gitRepoBrief     :: Text
-  , gitRepoEncrypted :: Bool
-  }
-  deriving stock (Generic,Data)
-
-
-data GitRepoFacts =
-      GitRepoRefFact  GitRepoRefFact
-    | GitRepoHeadFact HashRef GitRepoHeadFact
-    deriving stock (Generic,Data)
-
-
-instance Serialise GitRepoRefFact
-instance Serialise GitRepoHeadFact
-instance Serialise GitRepoFacts
-
-instance Pretty GitRepoFacts  where
-  pretty (GitRepoRefFact x)    = pretty x
-  pretty (GitRepoHeadFact ha x) = pretty ("gitrpoheadfact",ha,x)
-
-instance Pretty GitRepoRefFact where
-  pretty (GitRepoFact1{..}) =
-    parens ( "gitrepofact1" <+>  hsep [pretty gitLwwRef, pretty gitLwwSeq, pretty gitRefLog])
-
-instance Pretty GitRepoHeadFact where
-  pretty (GitRepoHeadFact1{..}) =
-    parens ( "gitrepoheadfact1" <+>  hsep [pretty gitRepoHeadRef])
-
-
 runOracleIndex :: forall m  . MonadUnliftIO m
                => PubKey 'Sign HBS2Basic
                -> Oracle m ()
@@ -112,6 +66,8 @@ runOracleIndex auPk = do
                            (lwwSeq lw)
                            (RefLogKey rk)
 
+  db <- asks _db
+
   facts <- S.toList_ do
 
     for_ repos $ \what@GitRepoFact1{..} -> do
@@ -128,6 +84,8 @@ runOracleIndex auPk = do
 
                     Right hxs -> do
                       for_ hxs $ \htx -> void $ runMaybeT do
+                        done <- liftIO $ withDB db (isTxProcessed (HashVal htx))
+                        guard (not done)
                         getBlock sto (fromHashRef htx) >>= toMPlus
                            <&> deserialiseOrFail @(RefLogUpdate L4Proto)
                            >>= toMPlus
@@ -137,6 +95,10 @@ runOracleIndex auPk = do
         let tx' = maximumByMay (comparing fst) txs
 
         for_ tx' $ \(n,tx) -> void $ runMaybeT do
+          liftIO $ withDB db do
+            transactional do
+              for_ [ t | (i,t) <- txs, i < n ] $ \tran -> do
+                insertTxProcessed (HashVal tran)
 
           (rhh,RepoHeadSimple{..}) <- readRepoHeadFromTx sto tx
                                          >>= toMPlus
@@ -152,9 +114,15 @@ runOracleIndex auPk = do
           let f2 = GitRepoHeadFact
                       repoFactHash
                       (GitRepoHeadFact1 rhh name brief enc)
+          let f3 = GitRepoHeadVersionFact repoFactHash (GitRepoHeadVersionFact1 _repoHeadTime)
+          let f4 = GitRepoTxFact gitLwwRef tx
 
           lift $ S.yield f1
           lift $ S.yield f2
+          lift $ S.yield f3
+          lift $ S.yield f4
+
+          liftIO $ withDB db (insertTxProcessed (HashVal tx))
 
   rchanAPI <- asks _refchanAPI
   chan  <- asks _refchanId
@@ -170,6 +138,10 @@ runOracleIndex auPk = do
     void $ callRpcWaitMay @RpcRefChanPropose (TimeoutSec 1) rchanAPI (chan, box)
     debug $ "posted tx" <+> pretty (hashObject @HbSync (serialise f))
 
+  -- FIXME: ASAP-wait-refchan-actually-updated
+  pause @'Seconds 0.25
+
+  updateState
 
 runDump :: forall m  . MonadUnliftIO m
         => PKS
@@ -178,14 +150,10 @@ runDump :: forall m  . MonadUnliftIO m
 runDump pks = do
   self <- liftIO getProgName
 
-  debug $ "fucking dump!" <+> pretty self
-
   let cmd = proc self ["pipe", "-r", show (pretty (AsBase58 pks))]
               & setStdin createPipe
               & setStdout createPipe
 
-  -- let w
-
   flip runContT pure do
 
     -- p <- ContT $ withProcessWait cmd
@@ -235,8 +203,6 @@ instance (MonadUnliftIO m, HasOracleEnv m) => HandleMethod m RpcChannelQuery whe
       rv <- lift (callRpcWaitMay @RpcRefChanGet (TimeoutSec 1) rchanAPI chan)
               >>= toMPlus >>= toMPlus
 
-      debug $ "AAAAAA" <+> pretty rv
-
       facts <- S.toList_ do
         walkMerkle @[HashRef] (fromHashRef rv) (getBlock sto) $ \case
           Left{} -> pure ()
@@ -271,13 +237,14 @@ instance (MonadUnliftIO m, HasOracleEnv m) => HandleMethod m RpcChannelQuery whe
         let nm    = maybe "" gitRepoName d
         let brief = maybe "" gitRepoBrief d
 
-        S.yield $ object [ "item_id"    .= show (pretty gitLwwRef)
+        S.yield $ Aeson.toJSON [ show (pretty gitLwwRef)
-                         , "item_title" .= show (pretty nm)
+                               , show (pretty nm)
-                         , "item_brief" .= show (pretty brief)
+                               , show (pretty brief)
-                         ]
+                               ]
 
-      let root = object [ "items" .= items
+      let root = object [ "rows"  .= items
                         , "state" .= show (pretty rv)
+                        , "desc"  .= [ "entity", "name", "brief" ]
                         ]
 
       pure $ A.encodePretty root
@@ -306,11 +273,6 @@ runPipe = do
   chan <- asks _refchanId
   debug "run pipe"
 
-  -- liftIO $ hSetBuffering stdin NoBuffering
-
-  -- liftIO $ LBS.getContents >>= LBS.hPutStr stderr
-  -- forever (pause @'Seconds 10)
-
   flip runContT pure do
 
     server  <- newMessagingPipe (stdin,stdout)
@@ -324,3 +286,73 @@ runPipe = do
           [ makeResponse (makeServer @BrowserPluginAPI)
           ]
 
+
+updateState :: MonadUnliftIO m => Oracle m ()
+updateState = do
+  debug $ yellow "update state"
+
+  chan  <- asks _refchanId
+  rchanAPI <- asks _refchanAPI
+  sto  <- asks  _storage
+
+  void $ runMaybeT do
+
+    rv <- lift (callRpcWaitMay @RpcRefChanGet (TimeoutSec 1) rchanAPI chan)
+            >>= toMPlus >>= toMPlus
+
+    facts <- S.toList_ do
+      walkMerkle @[HashRef] (fromHashRef rv) (getBlock sto) $ \case
+        Left{} -> pure ()
+        Right txs -> do
+          -- FIXME: skip-already-processed-blocks
+          for_ txs $ \htx -> void $ runMaybeT do
+            getBlock sto (fromHashRef htx)
+             >>= toMPlus
+             <&> deserialiseOrFail @(RefChanUpdate L4Proto)
+             >>= toMPlus
+             >>= \case
+                   Propose _ box -> pure box
+                   _             -> mzero
+             <&> unboxSignedBox0
+             >>= toMPlus
+             <&> snd
+             >>= \(ProposeTran _ box) -> toMPlus (unboxSignedBox0 box)
+             <&> snd
+             <&> deserialiseOrFail @GitRepoFacts . LBS.fromStrict
+             >>= toMPlus
+             >>= lift . S.yield
+
+    let rf  = [ (HashRef (hashObject $ serialise f), f)
+              | f@GitRepoFact1{} <-  universeBi facts
+              ] & HM.fromListWith (\v1 v2 -> if gitLwwSeq v1 > gitLwwSeq v2 then v1 else v2)
+
+
+    let rhf = [ (h,f) | (GitRepoHeadFact h f) <- universeBi facts ]
+              & HM.fromList
+
+    let rhtf  = [ (h,f) | (GitRepoHeadVersionFact h f) <- universeBi facts ]
+
+    let done =  [ (r,t) | GitRepoTxFact r t <- universeBi facts ]
+
+    lift $ withState do
+
+      transactional do
+
+        for_ done $ \(r,t) -> do
+          debug $ red "DONE" <+> pretty (r,t)
+
+        for_ (HM.toList rf) $ \(k, GitRepoFact1{..}) -> do
+
+          insertGitRepo (GitRepoKey gitLwwRef)
+
+          void $ runMaybeT do
+            d <- HM.lookup k rhf & toMPlus
+            lift do
+              insertGitRepoName (GitRepoKey gitLwwRef, HashVal k) (gitRepoName d)
+              insertGitRepoBrief(GitRepoKey gitLwwRef, HashVal k) (gitRepoBrief d)
+
+            pure ()
+
+        for_ rhtf  $ \(k, GitRepoHeadVersionFact1 v) -> do
+          insertGitRepoHeadVersion (HashVal k) v
+

hbs2-git/hbs2-git-oracle/lib/HBS2/Git/Oracle/State.hs

@@ -0,0 +1,131 @@
+module HBS2.Git.Oracle.State where
+
+import HBS2.Git.Oracle.Prelude
+import DBPipe.SQLite
+
+import Data.Coerce
+import Text.InterpolatedString.Perl6 (qc)
+import Data.Word
+
+evolveDB :: MonadUnliftIO m => DBPipeM m ()
+evolveDB = do
+  debug $ yellow "evolveDB"
+  gitRepoTable
+  gitRepoNameTable
+  gitRepoBriefTable
+  gitRepoHeadVersionTable
+  txProcessedTable
+
+txProcessedTable :: MonadUnliftIO m => DBPipeM m ()
+txProcessedTable = do
+  ddl [qc|
+    create table if not exists txprocessed
+    ( hash text not null primary key
+    )
+  |]
+
+gitRepoTable :: MonadUnliftIO m => DBPipeM m ()
+gitRepoTable = do
+  ddl [qc|
+    create table if not exists gitrepo
+    ( ref text not null primary key
+    )
+  |]
+
+
+gitRepoNameTable :: MonadUnliftIO m => DBPipeM m ()
+gitRepoNameTable = do
+  ddl [qc|
+    create table if not exists gitreponame
+    ( ref text not null
+    , hash text not null
+    , name text not null
+    , primary key (ref, hash)
+    )
+  |]
+
+gitRepoBriefTable :: MonadUnliftIO m => DBPipeM m ()
+gitRepoBriefTable = do
+  ddl [qc|
+    create table if not exists gitrepobrief
+    ( ref text not null
+    , hash text not null
+    , brief text not null
+    , primary key (ref, hash)
+    )
+  |]
+
+gitRepoHeadVersionTable :: MonadUnliftIO m => DBPipeM m ()
+gitRepoHeadVersionTable = do
+  ddl [qc|
+    create table if not exists gitrepoheadversion
+    ( hash text not null
+    , version integer not null
+    , primary key (hash)
+    )
+  |]
+
+newtype GitRepoKey = GitRepoKey (LWWRefKey HBS2Basic)
+                     deriving stock Generic
+
+newtype HashVal    = HashVal HashRef
+                     deriving stock Generic
+
+instance ToField GitRepoKey where
+  toField (GitRepoKey r) = toField $ show $ pretty $ AsBase58 r
+
+instance ToField HashVal where
+  toField (HashVal r) = toField $ show $ pretty $ AsBase58 r
+
+insertGitRepo :: MonadUnliftIO m => GitRepoKey -> DBPipeM m ()
+insertGitRepo repo = do
+  insert [qc|
+    insert into gitrepo(ref) values(?)
+    on conflict (ref) do nothing
+  |] (Only repo)
+
+insertGitRepoName :: MonadUnliftIO m
+                  => (GitRepoKey, HashVal)
+                  -> Text
+                  -> DBPipeM m ()
+insertGitRepoName (r,h)  name = do
+  insert [qc|
+    insert into gitreponame (ref,hash,name) values(?,?,?)
+    on conflict (ref,hash) do update set name = excluded.name
+  |] (r,h,name)
+
+insertGitRepoBrief :: MonadUnliftIO m
+                   => (GitRepoKey, HashVal)
+                   -> Text
+                   -> DBPipeM m ()
+insertGitRepoBrief (r,h) b = do
+  insert [qc|
+    insert into gitrepobrief (ref,hash,brief) values(?,?,?)
+    on conflict (ref,hash) do update set brief = excluded.brief
+  |] (r,h,b)
+
+
+insertGitRepoHeadVersion :: MonadUnliftIO m => HashVal -> Word64 -> DBPipeM m ()
+insertGitRepoHeadVersion hashVal version = do
+  insert [qc|
+    insert into gitrepoheadversion (hash, version) values(?,?)
+    on conflict (hash) do update set version = excluded.version
+  |] (hashVal, version)
+
+
+insertTxProcessed :: MonadUnliftIO m => HashVal -> DBPipeM m ()
+insertTxProcessed hash = do
+  insert [qc|
+    insert into txprocessed (hash) values (?)
+    on conflict do nothing
+  |] (Only hash)
+
+
+isTxProcessed :: MonadUnliftIO m => HashVal -> DBPipeM m Bool
+isTxProcessed hash = do
+  results <- select [qc|
+    select 1 from txprocessed where hash = ?
+  |] (Only hash)
+  pure $ not $ null (results :: [Only Int])
+
+

hbs2-git/hbs2-git-oracle/repo.sql

@@ -0,0 +1,18 @@
+
+SELECT
+  g.ref,
+  gn.name,
+  MAX(ghv.version) AS max_version,
+  gb.brief
+FROM
+  gitrepo AS g
+INNER JOIN
+  gitreponame AS gn ON g.ref = gn.ref
+INNER JOIN
+  gitrepoheadversion AS ghv ON gn.hash = ghv.hash
+LEFT JOIN
+  gitrepobrief AS gb ON g.ref = gb.ref AND ghv.hash = gb.hash
+GROUP BY
+  g.ref, gn.name;
+
+

hbs2-git/hbs2-git.cabal

@@ -176,6 +176,8 @@ library hbs2-git-oracle-oracle-lib
       HBS2.Git.Oracle.Prelude
       HBS2.Git.Oracle.App
       HBS2.Git.Oracle.Run
+      HBS2.Git.Oracle.State
+      HBS2.Git.Oracle.Facts
 
     build-depends:    base, hbs2-git
                     , base16-bytestring