fix review comments for offchain data

Author: Cmdv

cardano-db-sync/app/http-get-json-metadata.hs

@@ -4,10 +4,9 @@ import Cardano.Db (PoolMetaHash (..), PoolUrl (..))
 import Cardano.DbSync (
   OffChainFetchError (..),
   SimplifiedOffChainPoolData (..),
-  httpGetOffChainPoolData,
-  parseOffChainPoolUrl,
  )
 import Cardano.DbSync.Error (runOrThrowIO)
+import Cardano.DbSync.OffChain.Http (httpGetOffChainPoolData, parseOffChainPoolUrl)
 import Cardano.DbSync.Types (OffChainHashType (..), OffChainUrlType (..))
 import Control.Monad.IO.Class (liftIO)
 import Control.Monad.Trans.Except (ExceptT, runExceptT)

cardano-db-sync/src/Cardano/DbSync.hs

@@ -21,10 +21,6 @@ module Cardano.DbSync (
   -- For testing and debugging
   OffChainFetchError (..),
   SimplifiedOffChainPoolData (..),
-  httpGetOffChainPoolData,
-  parseOffChainPoolUrl,
-  httpGetOffChainVoteData,
-  parseOffChainVoteUrl,
 ) where
 
 import Cardano.BM.Trace (Trace, logError, logInfo, logWarning)
@@ -51,12 +47,6 @@ import Cardano.DbSync.Era
 import Cardano.DbSync.Error (SyncNodeError, hasAbortOnPanicEnv, runOrThrowIO)
 import Cardano.DbSync.Ledger.State
 import Cardano.DbSync.OffChain (runFetchOffChainPoolThread, runFetchOffChainVoteThread)
-import Cardano.DbSync.OffChain.Http (
-  httpGetOffChainPoolData,
-  httpGetOffChainVoteData,
-  parseOffChainPoolUrl,
-  parseOffChainVoteUrl,
- )
 import Cardano.DbSync.Rollback (unsafeRollback)
 import Cardano.DbSync.Sync (runSyncNodeClient)
 import Cardano.DbSync.Tracing.ToObjectOrphans ()

cardano-db-sync/src/Cardano/DbSync/Era/Shelley/Insert.hs

@@ -118,9 +118,8 @@ insertShelleyBlock syncEnv shouldLog withinTwoMins withinHalfHour blk details is
 
     slid <- lift . DB.insertSlotLeader $ Generic.mkSlotLeader (ioShelley iopts) (Generic.unKeyHashRaw $ Generic.blkSlotLeader blk) (eitherToMaybe mPhid)
     blkId <-
-      lift
-        . insertBlockAndCache cache
-        $ DB.Block
+      lift . insertBlockAndCache cache $
+        DB.Block
           { DB.blockHash = Generic.blkHash blk
           , DB.blockEpochNo = Just $ unEpochNo epochNo
           , DB.blockSlotNo = Just $ unSlotNo (Generic.blkSlotNo blk)
@@ -197,7 +196,7 @@ insertShelleyBlock syncEnv shouldLog withinTwoMins withinHalfHour blk details is
 
     insertStakeSlice syncEnv $ apStakeSlice applyResult
 
-    unless (ioGov iopts)
+    when (ioGov iopts)
       . lift
       $ insertOffChainVoteResults tracer (envOffChainVoteResultQueue syncEnv)
 

cardano-db-sync/src/Cardano/DbSync/OffChain.hs

@@ -84,8 +84,8 @@ loadOffChainWorkQueue ::
 loadOffChainWorkQueue _trce offChainWorkQueue = do
   whenM (liftIO $ atomically (isEmptyTBQueue (lQueue offChainWorkQueue))) $ do
     now <- liftIO Time.getPOSIXTime
-    runnablePools <- filter (isRunnable now offChainWorkQueue) <$> lGetData offChainWorkQueue now 100
-    liftIO $ mapM_ (queueInsert offChainWorkQueue) runnablePools
+    runnableOffChainPools <- filter (isRunnable now offChainWorkQueue) <$> lGetData offChainWorkQueue now 100
+    liftIO $ mapM_ (queueInsert offChainWorkQueue) runnableOffChainPools
   where
     isRunnable :: POSIXTime -> LoadOffChainWorkQueue a m -> a -> Bool
     isRunnable now oCWorkQueue locWq = retryRetryTime (lRetryTime oCWorkQueue locWq) <= now
@@ -103,13 +103,11 @@ insertOffChainPoolResults ::
   ReaderT SqlBackend m ()
 insertOffChainPoolResults trce resultQueue = do
   res <- liftIO . atomically $ flushTBQueue resultQueue
-  let resLength = length res
-      resErrorsLength = length $ filter isFetchError res
-  unless (null res)
-    $ liftIO
-      . logInfo trce
-    $ logInsertOffChainResults "Pool" resLength resErrorsLength
-
+  unless (null res) $ do
+    let resLength = length res
+        resErrorsLength = length $ filter isFetchError res
+    liftIO . logInfo trce $
+      logInsertOffChainResults "Pool" resLength resErrorsLength
   mapM_ insert res
   where
     insert :: (MonadBaseControl IO m, MonadIO m) => OffChainPoolResult -> ReaderT SqlBackend m ()
@@ -129,12 +127,11 @@ insertOffChainVoteResults ::
   ReaderT SqlBackend m ()
 insertOffChainVoteResults trce resultQueue = do
   res <- liftIO . atomically $ flushTBQueue resultQueue
-  let resLength = length res
-      resErrorsLength = length $ filter isFetchError res
-  unless (null res)
-    $ liftIO
-      . logInfo trce
-    $ logInsertOffChainResults "Voting Anchor" resLength resErrorsLength
+  unless (null res) $ do
+    let resLength = length res
+        resErrorsLength = length $ filter isFetchError res
+    liftIO . logInfo trce $
+      logInsertOffChainResults "Voting Anchor" resLength resErrorsLength
   mapM_ insert res
   where
     insert :: (MonadBaseControl IO m, MonadIO m) => OffChainVoteResult -> ReaderT SqlBackend m ()
@@ -174,7 +171,7 @@ runFetchOffChainPoolThread syncEnv = do
     forever $ do
       tDelay
       -- load the offChain vote work queue using the db
-      _ <- runReaderT (loadOffChainVoteWorkQueue trce (envOffChainVoteWorkQueue syncEnv)) (envBackend syncEnv)
+      _ <- runReaderT (loadOffChainPoolWorkQueue trce (envOffChainPoolWorkQueue syncEnv)) (envBackend syncEnv)
       poolq <- blockingFlushTBQueue (envOffChainPoolWorkQueue syncEnv)
       manager <- Http.newManager tlsManagerSettings
       now <- liftIO Time.getPOSIXTime
@@ -189,12 +186,12 @@ runFetchOffChainPoolThread syncEnv = do
 runFetchOffChainVoteThread :: SyncEnv -> IO ()
 runFetchOffChainVoteThread syncEnv = do
   -- if dissable gov is active then don't run voting anchor thread
-  unless (ioGov iopts) $ do
+  when (ioGov iopts) $ do
     logInfo trce "Running Offchain Vote Anchor fetch thread"
     forever $ do
       tDelay
       -- load the offChain pool work queue using the db
-      _ <- runReaderT (loadOffChainPoolWorkQueue trce (envOffChainPoolWorkQueue syncEnv)) (envBackend syncEnv)
+      _ <- runReaderT (loadOffChainVoteWorkQueue trce (envOffChainVoteWorkQueue syncEnv)) (envBackend syncEnv)
       voteq <- blockingFlushTBQueue (envOffChainVoteWorkQueue syncEnv)
       manager <- Http.newManager tlsManagerSettings
       now <- liftIO Time.getPOSIXTime

cardano-db-sync/src/Cardano/DbSync/OffChain/Http.hs

@@ -147,13 +147,12 @@ httpGetOffChain manager request mHash url =
           . left
           $ OCFErrHashMismatch url (renderByteArray eMetaHash) (renderByteArray metadataHash)
 
-      decodedMetadata <-
-        case Aeson.eitherDecode' respLBS of
-          Left err -> left $ OCFErrJsonDecodeFail url (Text.pack err)
-          Right res -> pure res
-
       case url of
-        OffChainPoolUrl _ ->
+        OffChainPoolUrl _ -> do
+          decodedMetadata <-
+            case Aeson.eitherDecode' respLBS of
+              Left err -> left $ OCFErrJsonDecodeFail url (Text.pack err)
+              Right res -> pure res
           pure $
             SimplifiedOffChainPoolDataType
               SimplifiedOffChainPoolData
@@ -172,10 +171,8 @@ httpGetOffChain manager request mHash url =
               SimplifiedOffChainVoteData
                 { sovaHash = metadataHash
                 , sovaBytes = respBS
-                , -- Instead of inserting the `respBS` here, we encode the JSON and then store that.
-                  -- This is necessary because the PostgreSQL JSON parser can reject some ByteStrings
-                  -- that the Aeson parser accepts.
-                  sovaJson = Text.decodeUtf8 $ LBS.toStrict (Aeson.encode decodedMetadata)
+                , -- TODO: no json format decided for vote metadata yet so we are leaving it blank for now
+                  sovaJson = ""
                 , sovaContentType = mContentType
                 }
 

cardano-db/src/Cardano/Db/Insert.hs

@@ -316,10 +316,14 @@ insertCheckOffChainPoolFetchError pofe = do
   when (foundPool && foundMeta) . void $ insertCheckUnique "OffChainPoolFetchError" pofe
 
 insertOffChainVoteData :: (MonadBaseControl IO m, MonadIO m) => OffChainVoteData -> ReaderT SqlBackend m ()
-insertOffChainVoteData pod = void $ insertCheckUnique "OffChainVoteData" pod
+insertOffChainVoteData ocvd = do
+  foundVotingAnchor <- existsVotingAnchorId (offChainVoteDataVotingAnchorId ocvd)
+  when foundVotingAnchor . void $ insertCheckUnique "OffChainVoteData" ocvd
 
 insertOffChainVoteFetchError :: (MonadBaseControl IO m, MonadIO m) => OffChainVoteFetchError -> ReaderT SqlBackend m ()
-insertOffChainVoteFetchError pofe = void $ insertCheckUnique "OffChainVoteFetchError" pofe
+insertOffChainVoteFetchError ocvfe = do
+  foundVotingAnchor <- existsVotingAnchorId (offChainVoteFetchErrorVotingAnchorId ocvfe)
+  when foundVotingAnchor . void $ insertCheckUnique "OffChainVoteFetchError" ocvfe
 
 insertReservedPoolTicker :: (MonadBaseControl IO m, MonadIO m) => ReservedPoolTicker -> ReaderT SqlBackend m (Maybe ReservedPoolTickerId)
 insertReservedPoolTicker ticker = do

cardano-db/src/Cardano/Db/Query.hs

@@ -53,6 +53,7 @@ module Cardano.Db.Query (
   queryMaxRefId,
   existsPoolHashId,
   existsPoolMetadataRefId,
+  existsVotingAnchorId,
   queryAdaPotsId,
   queryBlockHeight,
   queryAllExtraMigrations,
@@ -741,6 +742,15 @@ existsPoolMetadataRefId pmrid = do
     pure (pmr ^. PoolMetadataRefId)
   pure $ not (null res)
 
+existsVotingAnchorId :: MonadIO m => VotingAnchorId -> ReaderT SqlBackend m Bool
+existsVotingAnchorId vaId = do
+  res <- select $ do
+    votingAnchor <- from $ table @VotingAnchor
+    where_ (votingAnchor ^. VotingAnchorId ==. val vaId)
+    limit 1
+    pure (votingAnchor ^. VotingAnchorId)
+  pure $ not (null res)
+
 queryAdaPotsId :: MonadIO m => BlockId -> ReaderT SqlBackend m (Maybe (Entity AdaPots))
 queryAdaPotsId blkId = do
   res <- select $ do

cardano-db/src/Cardano/Db/Types.hs

@@ -238,7 +238,7 @@ newtype VoteUrl = VoteUrl {unVoteUrl :: Text}
   deriving (Eq, Ord, Generic)
   deriving (Show) via (Quiet VoteUrl)
 
--- | The raw binary hash of a stake vote metadata.
+-- | The raw binary hash of a vote metadata.
 newtype VoteMetaHash = VoteMetaHash {unVoteMetaHash :: ByteString}
   deriving (Eq, Ord, Generic)
   deriving (Show) via (Quiet VoteMetaHash)