Framework for Notifications

.gitignore

@@ -29,3 +29,4 @@ uniworx.nix
 src/Handler/Assist.bak
 src/Handler/Course.SnapCustom.hs
 *.orig
+/instance

config/settings.yml

@@ -41,6 +41,7 @@ ldap:
 userDefaultFavourites: 12
 
 cryptoid-keyfile: "_env:CRYPTOID_KEYFILE:cryptoid_key.bf"
+instance-idfile: "_env:INSTANCEID_FILE:instance"
 
 copyright: ©Institute for Informatics, LMU Munich
 #analytics: UA-YOURCODE

models

@@ -212,3 +212,10 @@ Exam
 --    -- CONTINUE HERE: Include rating in this table or separately?
 --    UniqueExamUser userId examId
 -- By default this file is used in Model.hs (which is imported by Foundation.hs)
+QueuedNotification
+    recipient           UserId
+    content             Value
+    created             UTCTime
+    lockInstance        UUID Maybe
+    lockTime            UTCTime Maybe
+    deriving Eq Read Show Generic Typeable

package.yaml

@@ -79,6 +79,7 @@ dependencies:
 - parsec
 - uuid
 - exceptions
+- stm-conduit
 
 # The library contains all of our application code. The executable
 # defined below is just a thin wrapper.

src/Application.hs

@@ -13,6 +13,7 @@ module Application
     , makeFoundation
     , makeLogWare
     -- * for DevelMain
+    , foundationStoreNum
     , getApplicationRepl
     , shutdownApp
     -- * for GHCI
@@ -37,6 +38,16 @@ import Network.Wai.Middleware.RequestLogger (Destination (Logger),
 import System.Log.FastLogger                (defaultBufSize, newStdoutLoggerSet,
                                              toLogStr)
 
+import Foreign.Store
+
+import qualified Data.UUID as UUID
+import qualified Data.UUID.V4 as UUID
+
+import System.Directory
+import System.FilePath
+
+import Notifications
+
 -- Import all relevant handler modules here.
 -- (HPack takes care to add new modules to our cabal file nowadays.)
 import Handler.Common
@@ -70,6 +81,12 @@ makeFoundation appSettings@(AppSettings{..}) = do
         appStaticDir
 
     appCryptoIDKey <- readKeyFile appCryptoIDKeyFile
+    appInstanceID <- maybe UUID.nextRandom readInstanceIDFile appInstanceIDFile
+
+    (appNotificationCtl, recvChan) <- atomically $ do
+        chan <- newBroadcastTMChan
+        recvChan <- dupTMChan chan
+        return (chan, recvChan)
 
     -- We need a log function to create a connection pool. We need a connection
     -- pool to create our foundation. And we need our foundation to get a
@@ -83,17 +100,35 @@ makeFoundation appSettings@(AppSettings{..}) = do
         tempFoundation = mkFoundation $ error "connPool forced in tempFoundation"
         logFunc = messageLoggerSource tempFoundation appLogger
 
+    flip runLoggingT logFunc . $(logDebugS) "InstanceID" $ UUID.toText appInstanceID
+
     -- Create the database connection pool
     pool <- flip runLoggingT logFunc $ createPostgresqlPool
         (pgConnStr  appDatabaseConf)
         (pgPoolSize appDatabaseConf)
-
+    
     -- Perform database migration using our application's logging settings.
-    runLoggingT (runSqlPool (runMigration migrateAll) pool) logFunc
+    flip runLoggingT logFunc $ runSqlPool (runMigration migrateAll) pool
+  
+    void . fork . handleNotifications $ (mkFoundation pool) { appNotificationCtl = recvChan }
 
     -- Return the foundation
     return $ mkFoundation pool
 
+readInstanceIDFile :: FilePath -> IO UUID
+readInstanceIDFile idFile = handle generateInstead $ readFileUtf8 idFile >>= parseText
+  where
+    parseText :: Text -> IO UUID
+    parseText = maybe (throwString "appInstanceIDFile does not contain an UUID") return . UUID.fromText
+    generateInstead :: IOException -> IO UUID
+    generateInstead e
+      | isDoesNotExistError e = do
+          createDirectoryIfMissing True $ takeDirectory idFile
+          instanceId <- UUID.nextRandom
+          writeFileUtf8 idFile $ UUID.toText instanceId
+          return instanceId
+      | otherwise = throw e
+
 -- | Convert our foundation to a WAI Application by calling @toWaiAppPlain@ and
 -- applying some additional middlewares.
 makeApplication :: UniWorX -> IO Application
@@ -172,16 +207,25 @@ appMain = do
 --------------------------------------------------------------
 -- Functions for DevelMain.hs (a way to run the app from GHCi)
 --------------------------------------------------------------
+foundationStoreNum :: Word32
+foundationStoreNum = 2
+  
 getApplicationRepl :: IO (Int, UniWorX, Application)
 getApplicationRepl = do
     settings <- getAppDevSettings
     foundation <- makeFoundation settings
     wsettings <- getDevSettings $ warpSettings foundation
     app1 <- makeApplication foundation
+
+    let foundationStore = Store foundationStoreNum
+    deleteStore foundationStore
+    writeStore foundationStore foundation
+  
     return (getPort wsettings, foundation, app1)
 
 shutdownApp :: UniWorX -> IO ()
-shutdownApp _ = return ()
+shutdownApp UniWorX{..} = do
+  atomically $ closeTMChan appNotificationCtl
 
 
 ---------------------------------------------

src/Foundation.hs

@@ -73,8 +73,13 @@ data UniWorX = UniWorX
     , appHttpManager :: Manager
     , appLogger      :: Logger
     , appCryptoIDKey :: CryptoIDKey
+    , appInstanceID  :: UUID
+    , appNotificationCtl :: TMChan NotificationCtl
     }
 
+data NotificationCtl = NCtlFlush
+                     | NCtlSend QueuedNotificationId
+
 -- This is where we define all of the routes in our application. For a full
 -- explanation of the syntax, please see:
 -- http://www.yesodweb.com/book/routing-and-handlers

src/Model.hs

@@ -7,6 +7,8 @@
 {-# LANGUAGE OverloadedStrings          #-}
 {-# LANGUAGE TemplateHaskell            #-}
 {-# LANGUAGE TypeFamilies               #-}
+{-# LANGUAGE DeriveDataTypeable         #-}
+{-# LANGUAGE DeriveGeneric              #-}
 module Model
   ( module Model
   , module Model.Types
@@ -17,6 +19,9 @@ import Database.Persist.Quasi
 -- import Data.Time
 -- import Data.ByteString
 import Model.Types
+import Data.UUID
+import Data.Aeson (Value)
+import Data.Aeson.TH (deriveJSON, defaultOptions)
 
 -- You can define all of your database entities in the entities file.
 -- You can find more information on persistent and how to declare entities
@@ -25,5 +30,7 @@ import Model.Types
 share [mkPersist sqlSettings, mkDeleteCascade sqlSettings, mkMigrate "migrateAll"]
     $(persistFileWith lowerCaseSettings "models")
 
-
+data Notification = SubmissionRated SubmissionId UTCTime
+  deriving (Eq, Ord, Show, Read)
+deriveJSON defaultOptions ''Notification
 

src/Model/Types.hs

@@ -28,8 +28,10 @@ import Text.Read (readMaybe)
 -- import Data.CaseInsensitive (CI)
 import qualified Data.CaseInsensitive as CI
 
+import Data.UUID
+
 import Yesod.Core.Dispatch (PathPiece(..))
-import Data.Aeson (FromJSON(..), ToJSON(..), withText, Value(..))
+import Data.Aeson (FromJSON(..), ToJSON(..), withText, Value(..), encode, eitherDecode)
 import Data.Aeson.TH (deriveJSON, defaultOptions)
 
 import GHC.Generics (Generic)
@@ -158,3 +160,14 @@ time `withinTerm` term = timeYear `mod` 100 == termYear `mod` 100
 data StudyFieldType = FieldPrimary | FieldSecondary
   deriving (Eq, Ord, Enum, Show, Read, Bounded)
 derivePersistField "StudyFieldType"
+
+derivePersistField "UUID"
+
+instance PersistField Value where
+  toPersistValue = PersistDbSpecific . toStrict . encode
+
+  fromPersistValue (PersistDbSpecific t) = first pack . eitherDecode $ fromStrict t
+  fromPersistValue _ = Left "JSON values must be converted from PersistDbSpecific"
+
+instance PersistFieldSql Value where
+  sqlType _ = SqlOther "json"

src/Notifications.hs

@@ -0,0 +1,78 @@
+{-# LANGUAGE NoImplicitPrelude
+           , RecordWildCards
+           , TemplateHaskell
+           , OverloadedStrings
+           , FlexibleContexts
+           , ViewPatterns
+           , TypeFamilies
+           , DeriveGeneric
+           , DeriveDataTypeable
+  #-}
+
+module Notifications
+  ( handleNotifications
+  ) where
+
+import Import
+
+import Data.Conduit.TMChan
+import qualified Data.Conduit.List as C
+
+import Data.Aeson (fromJSON, Result(..))
+import Database.Persist.Sql (rawExecute, fromSqlKey)
+
+
+data NotificationQueueException = QNInvalid QueuedNotification
+                                | QNLocked QueuedNotificationId UUID UTCTime
+                                | QNNonexistant QueuedNotificationId
+  deriving (Read, Show, Eq, Generic, Typeable)
+
+instance Exception NotificationQueueException
+  
+
+handleNotifications :: UniWorX -> IO ()
+handleNotifications foundation@UniWorX{..} = unsafeHandler foundation . bracket_ logStart logStop . runConduit $ sourceTMChan appNotificationCtl .| handleNotifications'
+  where
+    logStart = $(logDebugS) "Notifications" "Started"
+    logStop = $(logDebugS) "Notifications" "Shutting down"
+
+handleNotifications' :: Sink NotificationCtl Handler ()
+handleNotifications' = C.mapM_ $ void . handleAny ($(logErrorS) "Notifications" . tshow) . handleCmd
+  where
+    handleQueueException :: MonadLogger m => NotificationQueueException -> m ()
+    handleQueueException (QNInvalid qn) = $(logWarnS) "Notifications" $ "Invalid QueuedNotification: " ++ tshow qn
+    handleQueueException (QNNonexistant qnId) = $(logInfoS) "Notifications" $ "Saw nonexistant queue id: " ++ tshow (fromSqlKey qnId)
+    handleQueueException (QNLocked qnId lInstance lTime)  = $(logDebugS) "Notifications" $ "Saw locked QueuedNotification: " ++ tshow (qnId, lInstance, lTime)
+    
+    handleCmd NCtlFlush = void . fork . runDB . runConduit $ selectKeys [] [ Asc QueuedNotificationCreated ] .| C.mapM_ cmdSend
+    handleCmd (NCtlSend qnId) = handle handleQueueException $ do
+      qn@QueuedNotification{..} <- qnLock qnId
+
+      let
+        content :: Notification
+        Success content = fromJSON queuedNotificationContent
+      
+      $(logDebugS) "Notifications" $ "Would send: " <> tshow (queuedNotificationRecipient, content) -- FIXME
+
+      runDB $ delete qnId
+
+
+
+qnLock :: QueuedNotificationId -> Handler QueuedNotification
+qnLock qnId = runDB $ do
+  rawExecute "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE" []
+  qn@QueuedNotification{..} <- maybe (throwM $ QNNonexistant qnId) return =<< get qnId
+  maybe (return ()) throwM $ QNLocked <$> pure qnId <*> queuedNotificationLockInstance <*> queuedNotificationLockTime
+  unless ((fromJSON queuedNotificationContent :: Result Notification) /= mempty) . throwM $ QNInvalid qn
+  instanceID <- getsYesod appInstanceID
+  now <- liftIO getCurrentTime
+  updateGet qnId [ QueuedNotificationLockInstance =. Just instanceID
+                 , QueuedNotificationLockTime =. Just now
+                 ]
+
+cmdSend :: ( MonadHandler m
+           , HandlerSite m ~ UniWorX
+           ) => QueuedNotificationId -> m ()
+cmdSend (NCtlSend -> cmd) = do
+  chan <- getsYesod appNotificationCtl 
+  liftIO . atomically $ writeTMChan chan cmd

src/Settings.hs

@@ -68,6 +68,7 @@ data AppSettings = AppSettings
     , appAnalytics              :: Maybe Text
     -- ^ Google Analytics code
     , appCryptoIDKeyFile        :: FilePath
+    , appInstanceIDFile         :: Maybe FilePath
 
     , appAuthDummyLogin         :: Bool
     -- ^ Indicate if auth dummy login should be enabled.
@@ -102,6 +103,7 @@ instance FromJSON AppSettings where
         appCopyright              <- o .:  "copyright"
         appAnalytics              <- o .:? "analytics"
         appCryptoIDKeyFile        <- o .:  "cryptoid-keyfile"
+        appInstanceIDFile         <- o .:? "instance-idfile"
 
         appAuthDummyLogin         <- o .:? "auth-dummy-login"      .!= defaultDev