src/Gargantext/Core/NodeStory.hs

   1 {-|
   2 Module      : Gargantext.Core.NodeStory
   3 Description : Node API generation
   4 Copyright   : (c) CNRS, 2017-Present
   5 License     : AGPL + CECILL v3
   6 Maintainer  : team@gargantext.org
   7 Stability   : experimental
   8 Portability : POSIX
   9
  10 A Node Story is a Map between NodeId and an Archive (with state,
  11 version and history) for that node.
  12
  13 Couple of words on how this is implemented.
  14
  15 First version used files which stored Archive for each NodeId in a
  16 separate .cbor file.
  17
  18 For performance reasons, it is rewritten to use the DB.
  19
  20 The table `node_stories` contains two columns: `node_id` and
  21 `archive`.
  22
  23 Next, it was observed that `a_history` in `Archive` takes much
  24 space. So a new table was created, `node_story_archive_history` with
  25 columns: `node_id`, `ngrams_type_id`, `patch`. This is because each
  26 history item is in fact a map from `NgramsType` to `NgramsTablePatch`
  27 (see the `NgramsStatePatch'` type).
  28
  29 Moreover, since in ~G.A.Ngrams.commitStatePatch~ we use current state
  30 only, with only recent history items, I concluded that it is not
  31 necessary to load whole history into memory. Instead, it is kept in DB
  32 (history is immutable) and only recent changes are added to
  33 `a_history`. Then that record is cleared whenever `Archive` is saved.
  34
  35 Please note that
  36
  37 TODO:
  38 - remove
  39 - filter
  40 - charger les listes
  41 -}
  42
  43 {-# OPTIONS_GHC -fno-warn-orphans #-}
  44 {-# LANGUAGE Arrows #-}
  45 {-# LANGUAGE ConstraintKinds   #-}
  46 {-# LANGUAGE QuasiQuotes #-}
  47 {-# LANGUAGE TemplateHaskell   #-}
  48
  49 module Gargantext.Core.NodeStory
  50   ( HasNodeStory
  51   , HasNodeStoryEnv
  52   , hasNodeStory
  53   , HasNodeStoryVar
  54   , hasNodeStoryVar
  55   , HasNodeStorySaver
  56   , hasNodeStorySaver
  57   , HasNodeStoryImmediateSaver
  58   , hasNodeStoryImmediateSaver
  59   , NodeStory(..)
  60   , NgramsStatePatch'
  61   , NodeListStory
  62   , initNodeListStoryMock
  63   , NodeStoryEnv(..)
  64   , initNodeStory
  65   , nse_getter
  66   , nse_saver
  67   , nse_saver_immediate
  68   , nse_var
  69   , unNodeStory
  70   , getNodeArchiveHistory
  71   , Archive(..)
  72   , initArchive
  73   , insertArchiveList
  74   , deleteArchiveList
  75   , updateArchiveList
  76   , a_history
  77   , a_state
  78   , a_version
  79   , nodeExists
  80   , runPGSQuery
  81   , runPGSAdvisoryLock
  82   , runPGSAdvisoryUnlock
  83   , runPGSAdvisoryXactLock
  84   , getNodesIdWithType
  85   , readNodeStoryEnv
  86   , upsertNodeStories
  87   , getNodeStory
  88   , nodeStoriesQuery
  89   , currentVersion )
  90 where
  91
  92 -- import Debug.Trace (traceShow)
  93 import Control.Debounce (mkDebounce, defaultDebounceSettings, debounceFreq, debounceAction)
  94 import Codec.Serialise.Class
  95 import Control.Concurrent (MVar(), newMVar, modifyMVar_)
  96 import Control.Exception (catch, throw, SomeException(..))
  97 import Control.Lens (makeLenses, Getter, (^.), (.~), (%~), _Just, at, traverse, view)
  98 import Control.Monad.Except
  99 import Control.Monad.Reader
 100 import Data.Aeson hiding ((.=), decode)
 101 import Data.ByteString.Char8 (hPutStrLn)
 102 import Data.Map.Strict (Map)
 103 import Data.Maybe (catMaybes)
 104 import Data.Monoid
 105 import Data.Pool (Pool, withResource)
 106 import Data.Semigroup
 107 import Database.PostgreSQL.Simple.SqlQQ (sql)
 108 import Database.PostgreSQL.Simple.FromField (FromField(fromField), fromJSONField)
 109 import Data.Profunctor.Product.TH (makeAdaptorAndInstance)
 110 import GHC.Generics (Generic)
 111 import Gargantext.API.Ngrams.Types
 112 import Gargantext.Core.Types (ListId, NodeId(..), NodeType)
 113 import Gargantext.Core.Utils.Prefix (unPrefix)
 114 import Gargantext.Database.Admin.Config (nodeTypeId)
 115 import Gargantext.Database.Prelude (CmdM', HasConnectionPool(..), HasConfig)
 116 import Gargantext.Database.Query.Table.Node.Error (HasNodeError())
 117 import Gargantext.Prelude
 118 import Opaleye (DefaultFromField(..), SqlJsonb, fromPGSFromField)
 119 import System.IO (stderr)
 120 import qualified Data.Map.Strict                        as Map
 121 import qualified Data.Map.Strict.Patch                  as PM
 122 import qualified Data.Set as Set
 123 import qualified Data.Text as Text
 124 import qualified Database.PostgreSQL.Simple as PGS
 125 import qualified Gargantext.Database.Query.Table.Ngrams as TableNgrams
 126
 127 ------------------------------------------------------------------------
 128 data NodeStoryEnv = NodeStoryEnv
 129   { _nse_var    :: !(MVar NodeListStory)
 130   , _nse_saver  :: !(IO ())
 131   , _nse_saver_immediate :: !(IO ())
 132   , _nse_getter :: [NodeId] -> IO (MVar NodeListStory)
 133   --, _nse_cleaner :: !(IO ()) -- every 12 hours: cleans the repos of unused NodeStories
 134   -- , _nse_lock  :: !FileLock -- TODO (it depends on the option: if with database or file only)
 135   }
 136   deriving (Generic)
 137
 138 type HasNodeStory env err m = ( CmdM' env err m
 139                               , MonadReader env m
 140                               , MonadError  err m
 141                               , HasNodeStoryEnv env
 142                               , HasConfig env
 143                               , HasConnectionPool env
 144                               , HasNodeError err
 145                               )
 146
 147 class (HasNodeStoryVar env, HasNodeStorySaver env)
 148   => HasNodeStoryEnv env where
 149     hasNodeStory :: Getter env NodeStoryEnv
 150
 151 class HasNodeStoryVar env where
 152   hasNodeStoryVar :: Getter env ([NodeId] -> IO (MVar NodeListStory))
 153
 154 class HasNodeStorySaver env where
 155   hasNodeStorySaver :: Getter env (IO ())
 156
 157 class HasNodeStoryImmediateSaver env where
 158   hasNodeStoryImmediateSaver :: Getter env (IO ())
 159
 160 ------------------------------------------------------------------------
 161
 162 {- | Node Story for each NodeType where the Key of the Map is NodeId
 163   TODO : generalize for any NodeType, let's start with NodeList which
 164   is implemented already
 165 -}
 166 newtype NodeStory s p = NodeStory { _unNodeStory :: Map NodeId (Archive s p) }
 167   deriving (Generic, Show)
 168
 169 instance (FromJSON s, FromJSON p) => FromJSON (NodeStory s p)
 170 instance (ToJSON s, ToJSON p) => ToJSON (NodeStory s p)
 171 instance (Serialise s, Serialise p) => Serialise (NodeStory s p)
 172
 173 data Archive s p = Archive
 174   { _a_version           :: !Version
 175   , _a_state             :: !s
 176   , _a_history           :: ![p]
 177     -- first patch in the list is the most recent
 178     -- We use `take` in `commitStatePatch`, that's why.
 179
 180     -- History is immutable, we just insert things on top of existing
 181     -- list.
 182
 183     -- We don't need to store the whole history in memory, this
 184     -- structure holds only recent history, the one that will be
 185     -- inserted to the DB.
 186   }
 187   deriving (Generic, Show)
 188
 189 instance (Serialise s, Serialise p) => Serialise (Archive s p)
 190
 191
 192 type NodeListStory     = NodeStory NgramsState' NgramsStatePatch'
 193
 194 type NgramsState'      = Map       TableNgrams.NgramsType NgramsTableMap
 195 type NgramsStatePatch' = PatchMap  TableNgrams.NgramsType NgramsTablePatch
 196 instance Serialise NgramsStatePatch'
 197 instance FromField (Archive NgramsState' NgramsStatePatch')
 198   where
 199     fromField = fromJSONField
 200 instance DefaultFromField SqlJsonb (Archive NgramsState' NgramsStatePatch')
 201   where
 202     defaultFromField = fromPGSFromField
 203
 204 -- | Combine `NgramsState'`. This is because the structure is (Map
 205 -- NgramsType (Map ...)) and the default `(<>)` operator is
 206 -- left-biased
 207 -- (https://hackage.haskell.org/package/containers-0.6.6/docs/Data-Map-Internal.html#v:union)
 208 combineState :: NgramsState' -> NgramsState' -> NgramsState'
 209 combineState = Map.unionWith (<>)
 210
 211 instance (Semigroup s, Semigroup p) => Semigroup (Archive s p) where
 212   (<>) (Archive { _a_history = p }) (Archive { _a_version = v'
 213                                              , _a_state = s'
 214                                              , _a_history = p' }) =
 215     Archive { _a_version = v'
 216             , _a_state = s'
 217             , _a_history = p' <> p }
 218 instance (Monoid s, Semigroup p) => Monoid (Archive s p) where
 219   mempty = Archive { _a_version = 0
 220                    , _a_state = mempty
 221                    , _a_history = [] }
 222 instance (FromJSON s, FromJSON p) => FromJSON (Archive s p) where
 223   parseJSON = genericParseJSON $ unPrefix "_a_"
 224 instance (ToJSON s, ToJSON p) => ToJSON (Archive s p) where
 225   toJSON     = genericToJSON     $ unPrefix "_a_"
 226   toEncoding = genericToEncoding $ unPrefix "_a_"
 227
 228 ------------------------------------------------------------------------
 229 initNodeStory :: (Monoid s, Semigroup p) => NodeId -> NodeStory s p
 230 initNodeStory ni = NodeStory $ Map.singleton ni initArchive
 231
 232 initArchive :: (Monoid s, Semigroup p) => Archive s p
 233 initArchive = mempty
 234
 235 initNodeListStoryMock :: NodeListStory
 236 initNodeListStoryMock = NodeStory $ Map.singleton nodeListId archive
 237   where
 238     nodeListId = 0
 239     archive = Archive { _a_version = 0
 240                       , _a_state = ngramsTableMap
 241                       , _a_history = [] }
 242     ngramsTableMap = Map.singleton TableNgrams.NgramsTerms
 243                    $ Map.fromList
 244                    [ (n ^. ne_ngrams, ngramsElementToRepo n)
 245                    | n <- mockTable ^. _NgramsTable
 246                    ]
 247
 248 ------------------------------------------------------------------------
 249
 250
 251 ------------------------------------------------------------------------
 252 -- | Lenses at the bottom of the file because Template Haskell would reorder order of execution in others cases
 253 makeLenses ''NodeStoryEnv
 254 makeLenses ''NodeStory
 255 makeLenses ''Archive
 256
 257 ----------------------------------------------------------------------
 258 data NodeStoryPoly nid v ngtid ngid nre =
 259   NodeStoryDB { node_id             :: nid
 260               , version             :: v
 261               , ngrams_type_id      :: ngtid
 262               , ngrams_id           :: ngid
 263               , ngrams_repo_element :: nre }
 264   deriving (Eq)
 265
 266 data NodeStoryArchivePoly nid a =
 267   NodeStoryArchiveDB { a_node_id :: nid
 268                      , archive :: a }
 269   deriving (Eq)
 270
 271 $(makeAdaptorAndInstance "pNodeStory" ''NodeStoryPoly)
 272 $(makeAdaptorAndInstance "pNodeArchiveStory" ''NodeStoryArchivePoly)
 273
 274 -- type NodeStoryWrite = NodeStoryPoly (Column SqlInt4) (Column SqlInt4) (Column SqlInt4) (Column SqlInt4) (Column SqlJsonb)
 275 -- type NodeStoryRead = NodeStoryPoly (Column SqlInt4) (Column SqlInt4) (Column SqlInt4) (Column SqlInt4) (Column SqlJsonb)
 276
 277 -- type NodeStoryArchiveWrite = NodeStoryArchivePoly (Column SqlInt4) (Column SqlJsonb)
 278 -- type NodeStoryArchiveRead = NodeStoryArchivePoly (Column SqlInt4) (Column SqlJsonb)
 279
 280 type ArchiveList = Archive NgramsState' NgramsStatePatch'
 281
 282 -- DB stuff
 283
 284 runPGSExecute :: (PGS.ToRow q) => PGS.Connection -> PGS.Query -> q -> IO Int64
 285 runPGSExecute c qs a = catch (PGS.execute c qs a) printError
 286   where
 287     printError (SomeException e) = do
 288       --q' <- PGS.formatQuery c qs a
 289       _ <- panic $ Text.pack $ show e
 290       throw (SomeException e)
 291
 292 runPGSExecuteMany :: (PGS.ToRow q) => PGS.Connection -> PGS.Query -> [q] -> IO Int64
 293 runPGSExecuteMany c qs a = catch (PGS.executeMany c qs a) printError
 294   where
 295     printError (SomeException e) = do
 296       --q' <- PGS.formatQuery c qs a
 297       _ <- panic $ Text.pack $ show e
 298       throw (SomeException e)
 299
 300 runPGSQuery :: (PGS.FromRow r, PGS.ToRow q) => PGS.Connection -> PGS.Query -> q -> IO [r]
 301 runPGSQuery c q a = catch (PGS.query c q a) printError
 302   where
 303     printError (SomeException e) = do
 304       q' <- PGS.formatQuery c q a
 305       hPutStrLn stderr q'
 306       throw (SomeException e)
 307
 308 runPGSAdvisoryLock :: PGS.Connection -> Int -> IO ()
 309 runPGSAdvisoryLock c id = do
 310   _ <- runPGSQuery c [sql| SELECT pg_advisory_lock(?) |] (PGS.Only id) :: IO [PGS.Only ()]
 311   pure ()
 312
 313 runPGSAdvisoryUnlock :: PGS.Connection -> Int -> IO ()
 314 runPGSAdvisoryUnlock c id = do
 315   _ <- runPGSQuery c [sql| SELECT pg_advisory_unlock(?) |] (PGS.Only id) :: IO [PGS.Only Bool]
 316   pure ()
 317
 318 runPGSAdvisoryXactLock :: PGS.Connection -> Int -> IO ()
 319 runPGSAdvisoryXactLock c id = do
 320   _ <- runPGSQuery c [sql| SELECT pg_advisory_xact_lock(?) |] (PGS.Only id) :: IO [PGS.Only ()]
 321   pure ()
 322
 323 nodeExists :: PGS.Connection -> NodeId -> IO Bool
 324 nodeExists c nId = (== [PGS.Only True])
 325   <$> runPGSQuery c [sql| SELECT true FROM nodes WHERE id = ? LIMIT 1 |] (PGS.Only nId)
 326
 327 getNodesIdWithType :: PGS.Connection -> NodeType -> IO [NodeId]
 328 getNodesIdWithType c nt = do
 329   ns <- runPGSQuery c query (PGS.Only $ nodeTypeId nt)
 330   pure $ map (\(PGS.Only nId) -> NodeId nId) ns
 331   where
 332     query :: PGS.Query
 333     query = [sql| SELECT id FROM nodes WHERE typename = ? |]
 334
 335
 336
 337 -- nodeStoryTable :: Table NodeStoryRead NodeStoryWrite
 338 -- nodeStoryTable =
 339 --   Table "node_stories"
 340 --     ( pNodeStory NodeStoryDB { node_id             = tableField "node_id"
 341 --                              , version             = tableField "version"
 342 --                              , ngrams_type_id      = tableField "ngrams_type_id"
 343 --                              , ngrams_id           = tableField "ngrams_id"
 344 --                              , ngrams_repo_element = tableField "ngrams_repo_element"
 345 --                              } )
 346
 347 -- nodeStoryArchiveTable :: Table NodeStoryArchiveRead NodeStoryArchiveWrite
 348 -- nodeStoryArchiveTable =
 349 --   Table "node_story_archive_history"
 350 --     ( pNodeArchiveStory NodeStoryArchiveDB { a_node_id = tableField "node_id"
 351 --                                            , archive   = tableField "archive" } )
 352
 353 -- nodeStorySelect :: Select NodeStoryRead
 354 -- nodeStorySelect = selectTable nodeStoryTable
 355
 356 -- NOTE "first patch in the _a_history list is the most recent"
 357 getNodeArchiveHistory :: PGS.Connection -> NodeId -> IO [NgramsStatePatch']
 358 getNodeArchiveHistory c nodeId = do
 359   as <- runPGSQuery c query (PGS.Only nodeId) :: IO [(TableNgrams.NgramsType, NgramsTerm, NgramsPatch)]
 360   pure $ (\(ngramsType, terms, patch) -> fst $ PM.singleton ngramsType (NgramsTablePatch $ fst $ PM.singleton terms patch)) <$> as
 361   where
 362     query :: PGS.Query
 363     query = [sql| SELECT ngrams_type_id, terms, patch
 364                     FROM node_story_archive_history
 365                     JOIN ngrams ON ngrams.id = ngrams_id
 366                     WHERE node_id = ?
 367                     ORDER BY (version, node_story_archive_history.id) DESC |]
 368
 369 ngramsIdQuery :: PGS.Query
 370 ngramsIdQuery = [sql| SELECT id FROM ngrams WHERE terms = ? |]
 371
 372
 373 insertNodeArchiveHistory :: PGS.Connection -> NodeId -> Version -> [NgramsStatePatch'] -> IO ()
 374 insertNodeArchiveHistory _ _ _ [] = pure ()
 375 insertNodeArchiveHistory c nodeId version (h:hs) = do
 376   let tuples = mconcat $ (\(nType, (NgramsTablePatch patch)) ->
 377                            (\(term, p) ->
 378                               (nodeId, nType, term, p)) <$> PM.toList patch) <$> PM.toList h :: [(NodeId, TableNgrams.NgramsType, NgramsTerm, NgramsPatch)]
 379   tuplesM <- mapM (\(nId, nType, term, patch) -> do
 380                       ngrams <- runPGSQuery c ngramsIdQuery (PGS.Only term)
 381                       pure $ (\(PGS.Only termId) -> (nId, nType, termId, term, patch)) <$> (headMay ngrams)
 382                       ) tuples :: IO [Maybe (NodeId, TableNgrams.NgramsType, Int, NgramsTerm, NgramsPatch)]
 383   _ <- runPGSExecuteMany c query $ ((\(nId, nType, termId, _term, patch) -> (nId, nType, termId, patch, version)) <$> (catMaybes tuplesM))
 384   _ <- insertNodeArchiveHistory c nodeId version hs
 385   pure ()
 386   where
 387
 388     query :: PGS.Query
 389     query = [sql| INSERT INTO node_story_archive_history(node_id, ngrams_type_id, ngrams_id, patch, version) VALUES (?, ?, ?, ?, ?) |]
 390
 391 getNodeStory :: PGS.Connection -> NodeId -> IO NodeListStory
 392 getNodeStory c nId@(NodeId nodeId) = do
 393   --res <- withResource pool $ \c -> runSelect c query :: IO [NodeStoryPoly NodeId Version Int Int NgramsRepoElement]
 394   res <- runPGSQuery c nodeStoriesQuery (PGS.Only nodeId) :: IO [(Version, TableNgrams.NgramsType, NgramsTerm, NgramsRepoElement)]
 395   -- We have multiple rows with same node_id and different (ngrams_type_id, ngrams_id).
 396   -- Need to create a map: {<node_id>: {<ngrams_type_id>: {<ngrams_id>: <data>}}}
 397   let dbData = map (\(version, ngramsType, ngrams, ngrams_repo_element) ->
 398                       Archive { _a_version = version
 399                               , _a_history = []
 400                               , _a_state   = Map.singleton ngramsType $ Map.singleton ngrams ngrams_repo_element }) res
 401   -- NOTE When concatenating, check that the same version is for all states
 402   pure $ NodeStory $ Map.singleton nId $ foldl combine mempty dbData
 403   --pure $ NodeStory $ Map.fromListWith (<>) $ (\(NodeStoryDB nId a) -> (nId, a)) <$> res
 404   where
 405     -- NOTE (<>) for Archive doesn't concatenate states, so we have to use `combine`
 406     combine a1 a2 = a1 & a_state %~ combineState (a2 ^. a_state)
 407                        & a_version .~ (a2 ^. a_version)  -- version should be updated from list, not taken from the empty Archive
 408
 409 nodeStoriesQuery :: PGS.Query
 410 nodeStoriesQuery = [sql| SELECT version, ngrams_type_id, terms, ngrams_repo_element
 411                            FROM node_stories
 412                            JOIN ngrams ON ngrams.id = ngrams_id
 413                            WHERE node_id = ? |]
 414
 415 type ArchiveStateList = [(TableNgrams.NgramsType, NgramsTerm, NgramsRepoElement)]
 416
 417 -- Functions to convert archive state (which is a Map NgramsType (Map
 418 -- NgramsTerm NgramsRepoElement)) to/from a flat list
 419 archiveStateAsList :: NgramsState' -> ArchiveStateList
 420 archiveStateAsList s = mconcat $ (\(nt, ntm) -> (\(n, nre) -> (nt, n, nre)) <$> Map.toList ntm) <$> Map.toList s
 421
 422 archiveStateFromList :: ArchiveStateList -> NgramsState'
 423 archiveStateFromList l = Map.fromListWith (<>) $ (\(nt, t, nre) -> (nt, Map.singleton t nre)) <$> l
 424
 425 -- | This function inserts whole new node story and archive for given node_id.
 426 insertNodeStory :: PGS.Connection -> NodeId -> ArchiveList -> IO ()
 427 insertNodeStory c (NodeId nId) a = do
 428   _ <- mapM (\(ngramsType, ngrams, ngramsRepoElement) -> do
 429                 termIdM <- runPGSQuery c ngramsIdQuery (PGS.Only ngrams) :: IO [PGS.Only Int64]
 430                 case headMay termIdM of
 431                   Nothing -> pure 0
 432                   Just (PGS.Only termId) -> runPGSExecuteMany c query [(nId, a ^. a_version, ngramsType, termId, ngramsRepoElement)]) $ archiveStateAsList $ a ^. a_state
 433              -- runInsert c $ insert ngramsType ngrams ngramsRepoElement) $ archiveStateAsList _a_state
 434
 435   pure ()
 436   where
 437     query :: PGS.Query
 438     query = [sql| INSERT INTO node_stories(node_id, ngrams_type_id, ngrams_id, ngrams_repo_element) VALUES (?, ?, ?, ?) |]
 439     -- insert ngramsType ngrams ngramsRepoElement =
 440     --   Insert { iTable      = nodeStoryTable
 441     --          , iRows       = [NodeStoryDB { node_id = sqlInt4 nId
 442     --                                       , version = sqlInt4 _a_version
 443     --                                       , ngrams_type_id = sqlInt4 $ TableNgrams.ngramsTypeId ngramsType
 444     --                                       , ngrams_id = ...
 445     --                                       , ngrams_repo_element = sqlValueJSONB ngramsRepoElement
 446     --                                       }]
 447     --          , iReturning  = rCount
 448     --          , iOnConflict = Nothing }
 449
 450 insertArchiveList :: PGS.Connection -> NodeId -> ArchiveList -> IO ()
 451 insertArchiveList c nodeId a = do
 452   _ <- mapM_ (\(nt, n, nre) -> runPGSExecute c query (nodeId, a ^. a_version, nt, nre, n)) (archiveStateAsList $ a ^. a_state)
 453   --_ <- runPGSExecuteMany c query $ (\(nt, n, nre) -> (nodeId, a ^. a_version, nt, nre, n)) <$> (archiveStateAsList $ a ^. a_state)
 454   pure ()
 455   where
 456     query :: PGS.Query
 457     query = [sql| WITH s as (SELECT ? as sid, ? sversion, ? sngrams_type_id, ngrams.id as sngrams_id, ?::jsonb as srepo FROM ngrams WHERE terms = ?)
 458                   INSERT INTO node_stories(node_id, version, ngrams_type_id, ngrams_id, ngrams_repo_element)
 459                   SELECT s.sid, s.sversion, s.sngrams_type_id, s.sngrams_id, s.srepo from s s join nodes n on s.sid = n.id
 460             |]
 461
 462 deleteArchiveList :: PGS.Connection -> NodeId -> ArchiveList -> IO ()
 463 deleteArchiveList c nodeId a = do
 464   _ <- mapM_ (\(nt, n, _) -> runPGSExecute c query (nodeId, nt, n)) (archiveStateAsList $ a ^. a_state)
 465   --_ <- runPGSExecuteMany c query $ (\(nt, n, _) -> (nodeId, nt, n)) <$> (archiveStateAsList $ a ^. a_state)
 466   pure ()
 467   where
 468     query :: PGS.Query
 469     query = [sql| DELETE FROM node_stories
 470                     WHERE node_id = ? AND ngrams_type_id = ? AND ngrams_id IN (SELECT id FROM ngrams WHERE terms = ?) |]
 471
 472 updateArchiveList :: PGS.Connection -> NodeId -> ArchiveList -> IO ()
 473 updateArchiveList c nodeId a = do
 474   let params = (\(nt, n, nre) -> (nre, nodeId, nt, n)) <$> (archiveStateAsList $ a ^. a_state)
 475   --q <- PGS.format c query params
 476   --printDebug "[updateArchiveList] query" q
 477   _ <- mapM (\p -> runPGSExecute c query p) params
 478   pure ()
 479   where
 480     query :: PGS.Query
 481     query = [sql| UPDATE node_stories
 482                     SET ngrams_repo_element = ?
 483                     WHERE node_id = ? AND ngrams_type_id = ? AND ngrams_id IN (SELECT id FROM ngrams WHERE terms = ?) |]
 484
 485 -- | This function updates the node story and archive for given node_id.
 486 updateNodeStory :: PGS.Connection -> NodeId -> ArchiveList -> ArchiveList -> IO ()
 487 updateNodeStory c nodeId@(NodeId _nId) currentArchive newArchive = do
 488   -- STEPS
 489
 490   -- 0. We assume we're inside an advisory lock
 491
 492   -- 1. Find differences (inserts/updates/deletes)
 493   let currentList = archiveStateAsList $ currentArchive ^. a_state
 494   let newList = archiveStateAsList $ newArchive ^. a_state
 495   let currentSet = Set.fromList $ (\(nt, n, _) -> (nt, n)) <$> currentList
 496   let newSet = Set.fromList $ (\(nt, n, _) -> (nt, n)) <$> newList
 497
 498   let inserts = filter (\(nt, n, _) -> Set.member (nt, n) $ Set.difference newSet currentSet) newList
 499   --printDebug "[updateNodeStory] inserts" inserts
 500   let deletes = filter (\(nt, n, _) -> Set.member (nt, n) $ Set.difference currentSet newSet) currentList
 501   --printDebug "[updateNodeStory] deletes" deletes
 502
 503   -- updates are the things that are in new but not in current
 504   let updates = Set.toList $ Set.difference (Set.fromList newList) (Set.fromList currentList)
 505   --printDebug "[updateNodeStory] updates" $ Text.unlines $ (Text.pack . show) <$> updates
 506
 507   -- 2. Perform inserts/deletes/updates
 508   --printDebug "[updateNodeStory] applying insert" ()
 509   insertArchiveList c nodeId $ Archive { _a_version = newArchive ^. a_version
 510                                        , _a_history = []
 511                                        , _a_state = archiveStateFromList inserts }
 512   --printDebug "[updateNodeStory] insert applied" ()
 513     --TODO Use currentArchive ^. a_version in delete and report error
 514   -- if entries with (node_id, ngrams_type_id, ngrams_id) but
 515   -- different version are found.
 516   deleteArchiveList c nodeId $ Archive { _a_version = newArchive ^. a_version
 517                                        , _a_history = []
 518                                        , _a_state = archiveStateFromList deletes }
 519   --printDebug "[updateNodeStory] delete applied" ()
 520   updateArchiveList c nodeId $ Archive { _a_version = newArchive ^. a_version
 521                                        , _a_history = []
 522                                        , _a_state = archiveStateFromList updates }
 523   --printDebug "[updateNodeStory] update applied" ()
 524
 525   pure ()
 526   -- where
 527   --   update = Update { uTable      = nodeStoryTable
 528   --                   , uUpdateWith = updateEasy (\(NodeStoryDB { node_id }) ->
 529   --                                                 NodeStoryDB { archive = sqlValueJSONB $ Archive { _a_history = emptyHistory
 530   --                                                                                                                               , ..}
 531   --                                                                                           , .. })
 532   --                   , uWhere      = (\row -> node_id row .== sqlInt4 nId)
 533   --                   , uReturning  = rCount }
 534
 535 -- nodeStoryRemove :: Pool PGS.Connection -> NodeId -> IO Int64
 536 -- nodeStoryRemove pool (NodeId nId) = withResource pool $ \c -> runDelete c delete
 537 --   where
 538 --     delete = Delete { dTable     = nodeStoryTable
 539 --                     , dWhere     = (\row -> node_id row .== sqlInt4 nId)
 540 --                     , dReturning = rCount }
 541
 542 upsertNodeStories :: PGS.Connection -> NodeId -> ArchiveList -> IO ()
 543 upsertNodeStories c nodeId@(NodeId nId) newArchive = do
 544   printDebug "[upsertNodeStories] START nId" nId
 545   PGS.withTransaction c $ do
 546     printDebug "[upsertNodeStories] locking nId" nId
 547     runPGSAdvisoryXactLock c nId
 548
 549     -- whether it's insert or update, we can insert node archive history already
 550     -- NOTE: It is assumed that the most recent change is the first in the
 551     -- list, so we save these in reverse order
 552     insertNodeArchiveHistory c nodeId (newArchive ^. a_version) $ reverse $ newArchive ^. a_history
 553
 554     (NodeStory m) <- getNodeStory c nodeId
 555     case Map.lookup nodeId m of
 556       Nothing -> do
 557         _ <- insertNodeStory c nodeId newArchive
 558         pure ()
 559       Just currentArchive  -> do
 560         _ <- updateNodeStory c nodeId currentArchive newArchive
 561         pure ()
 562
 563     printDebug "[upsertNodeStories] STOP nId" nId
 564
 565 writeNodeStories :: PGS.Connection -> NodeListStory -> IO ()
 566 writeNodeStories c (NodeStory nls) = do
 567   _ <- mapM (\(nId, a) -> upsertNodeStories c nId a) $ Map.toList nls
 568   pure ()
 569
 570 -- | Returns a `NodeListStory`, updating the given one for given `NodeId`
 571 nodeStoryInc :: PGS.Connection -> Maybe NodeListStory -> NodeId -> IO NodeListStory
 572 nodeStoryInc c Nothing nId = getNodeStory c nId
 573 nodeStoryInc c (Just ns@(NodeStory nls)) nId = do
 574   case Map.lookup nId nls of
 575     Nothing -> do
 576       (NodeStory nls') <- getNodeStory c nId
 577       pure $ NodeStory $ Map.union nls nls'
 578     Just _ -> pure ns
 579
 580 nodeStoryIncs :: PGS.Connection -> Maybe NodeListStory -> [NodeId] -> IO NodeListStory
 581 nodeStoryIncs _ Nothing [] = pure $ NodeStory $ Map.empty
 582 nodeStoryIncs c (Just nls) ns = foldM (\m n -> nodeStoryInc c (Just m) n) nls ns
 583 nodeStoryIncs c Nothing (ni:ns) = do
 584   m <- getNodeStory c ni
 585   nodeStoryIncs c (Just m) ns
 586
 587 -- nodeStoryDec :: Pool PGS.Connection -> NodeListStory -> NodeId -> IO NodeListStory
 588 -- nodeStoryDec pool ns@(NodeStory nls) ni = do
 589 --   case Map.lookup ni nls of
 590 --     Nothing -> do
 591 --       _ <- nodeStoryRemove pool ni
 592 --       pure ns
 593 --     Just _ -> do
 594 --       let ns' = Map.filterWithKey (\k _v -> k /= ni) nls
 595 --       _ <- nodeStoryRemove pool ni
 596 --       pure $ NodeStory ns'
 597 ------------------------------------
 598
 599 readNodeStoryEnv :: Pool PGS.Connection -> IO NodeStoryEnv
 600 readNodeStoryEnv pool = do
 601   mvar <- nodeStoryVar pool Nothing []
 602   saver <- mkNodeStorySaver pool mvar
 603   let saver_immediate = modifyMVar_ mvar $ \ns -> do
 604         withResource pool $ \c -> do
 605           --printDebug "[mkNodeStorySaver] will call writeNodeStories, ns" ns
 606           writeNodeStories c ns
 607           pure $ clearHistory ns
 608   -- let saver = modifyMVar_ mvar $ \mv -> do
 609   --       writeNodeStories pool mv
 610   --       printDebug "[readNodeStoryEnv] saver" mv
 611   --       let mv' = clearHistory mv
 612   --       printDebug "[readNodeStoryEnv] saver, cleared" mv'
 613   --       return mv'
 614   pure $ NodeStoryEnv { _nse_var    = mvar
 615                       , _nse_saver  = saver
 616                       , _nse_saver_immediate = saver_immediate
 617                       , _nse_getter = nodeStoryVar pool (Just mvar)
 618                       }
 619
 620 nodeStoryVar :: Pool PGS.Connection
 621              -> Maybe (MVar NodeListStory)
 622              -> [NodeId]
 623              -> IO (MVar NodeListStory)
 624 nodeStoryVar pool Nothing nIds = do
 625   state <- withResource pool $ \c -> nodeStoryIncs c Nothing nIds
 626   newMVar state
 627 nodeStoryVar pool (Just mv) nIds = do
 628   _ <- withResource pool
 629       $ \c -> modifyMVar_ mv
 630       $ \nsl -> (nodeStoryIncs c (Just nsl) nIds)
 631   pure mv
 632
 633 -- Debounce is useful since it could delay the saving to some later
 634 -- time, asynchronously and we keep operating on memory only.
 635 mkNodeStorySaver :: Pool PGS.Connection -> MVar NodeListStory -> IO (IO ())
 636 mkNodeStorySaver pool mvns = mkDebounce settings
 637   where
 638     settings = defaultDebounceSettings
 639                  { debounceAction = do
 640                      -- NOTE: Lock MVar first, then use resource pool.
 641                      -- Otherwise we could wait for MVar, while
 642                      -- blocking the pool connection.
 643                      modifyMVar_ mvns $ \ns -> do
 644                        withResource pool $ \c -> do
 645                          --printDebug "[mkNodeStorySaver] will call writeNodeStories, ns" ns
 646                          writeNodeStories c ns
 647                          pure $ clearHistory ns
 648                      --withMVar mvns (\ns -> printDebug "[mkNodeStorySaver] debounce nodestory" ns)
 649                  , debounceFreq = 1*minute
 650                  }
 651     minute = 60*second
 652     second = 10^(6 :: Int)
 653
 654 clearHistory :: NodeListStory -> NodeListStory
 655 clearHistory (NodeStory ns) = NodeStory $ ns & (traverse . a_history) .~ emptyHistory
 656   where
 657     emptyHistory = [] :: [NgramsStatePatch']
 658
 659 currentVersion :: (HasNodeStory env err m) => ListId -> m Version
 660 currentVersion listId = do
 661   pool <- view connPool
 662   nls <- withResource pool $ \c -> liftBase $ getNodeStory c listId
 663   pure $ nls ^. unNodeStory . at listId . _Just . a_version
 664
 665
 666 -- mkNodeStorySaver :: MVar NodeListStory -> Cmd err (Cmd err ())
 667 -- mkNodeStorySaver mvns = mkDebounce settings
 668 --   where
 669 --     settings = defaultDebounceSettings
 670 --                  { debounceAction = withMVar mvns (\ns -> writeNodeStories ns)
 671 --                  , debounceFreq = 1 * minute
 672 -- --                 , debounceEdge = trailingEdge -- Trigger on the trailing edge
 673 --                  }
 674 --     minute = 60 * second
 675 --     second = 10^(6 :: Int)
 676
 677
 678 -----------------------------------------