397 lines
18 KiB
Haskell
397 lines
18 KiB
Haskell
{-# LANGUAGE DataKinds #-}
|
|
{-# LANGUAGE FlexibleInstances #-}
|
|
{-# LANGUAGE InstanceSigs #-}
|
|
{-# LANGUAGE MultiParamTypeClasses #-}
|
|
{-# LANGUAGE OverloadedStrings #-}
|
|
{-# LANGUAGE RankNTypes #-}
|
|
{-# LANGUAGE TypeOperators #-}
|
|
|
|
module Hash2Pub.PostService where
|
|
|
|
import Control.Concurrent
|
|
import Control.Concurrent.Async
|
|
import Control.Concurrent.STM
|
|
import Control.Concurrent.STM.TChan
|
|
import Control.Concurrent.STM.TQueue
|
|
import Control.Concurrent.STM.TVar
|
|
import Control.Monad (foldM, forM_, forever)
|
|
import Control.Monad.IO.Class (liftIO)
|
|
import qualified Data.ByteString.Lazy.UTF8 as BSUL
|
|
import qualified Data.ByteString.UTF8 as BSU
|
|
import qualified Data.HashMap.Strict as HMap
|
|
import qualified Data.HashSet as HSet
|
|
import Data.Maybe (fromMaybe, isJust)
|
|
import Data.String (fromString)
|
|
import qualified Data.Text.Lazy as Txt
|
|
import Data.Text.Normalize (NormalizationMode (NFC),
|
|
normalize)
|
|
import Data.Time.Clock.POSIX
|
|
import Data.Typeable (Typeable)
|
|
import qualified Network.HTTP.Client as HTTP
|
|
import System.Random
|
|
import Text.Read (readEither)
|
|
|
|
import qualified Network.Wai.Handler.Warp as Warp
|
|
import Servant
|
|
import Servant.Client
|
|
import Servant.Server
|
|
|
|
import Hash2Pub.FediChordTypes
|
|
import Hash2Pub.RingMap
|
|
|
|
|
|
data PostService d = PostService
|
|
{ serviceConf :: ServiceConf
|
|
-- queues, other data structures
|
|
, baseDHT :: (DHT d) => d
|
|
, serviceThread :: TVar ThreadId
|
|
, subscribers :: TVar RelayTags
|
|
-- ^ for each tag store the subscribers + their queue
|
|
, ownSubscriptions :: TVar (HMap.HashMap NodeID POSIXTime)
|
|
-- ^ tags subscribed by the own node have an assigned lease time
|
|
, ownPosts :: TVar (HSet.HashSet Txt.Text)
|
|
-- ^ just store the existence of posts for saving memory,
|
|
, relayInQueue :: TQueue (Hashtag, PostID, PostContent)
|
|
-- ^ Queue for processing incoming posts of own instance asynchronously
|
|
, postFetchQueue :: TQueue PostID
|
|
}
|
|
deriving (Typeable)
|
|
|
|
type Hashtag = Txt.Text
|
|
type PostID = Txt.Text
|
|
type PostContent = Txt.Text
|
|
-- | For each handled tag, store its subscribers and provide a
|
|
-- broadcast 'TChan' for enqueuing posts
|
|
type RelayTags = RingMap NodeID (TagSubscribersSTM, TChan PostID, Hashtag)
|
|
type TagSubscribersSTM = TVar TagSubscribers
|
|
-- | each subscriber is identified by its contact data "hostname" "port"
|
|
-- and holds a TChan duplicated from the broadcast TChan of the tag
|
|
-- + an expiration timestamp
|
|
type TagSubscribers = (HMap.HashMap (String, Int) (TChan PostID, POSIXTime))
|
|
|
|
|
|
instance DHT d => Service PostService d where
|
|
-- | initialise 'PostService' data structures and run server
|
|
runService conf dht = do
|
|
-- create necessary TVars
|
|
threadVar <- newTVarIO =<< myThreadId -- own thread ID as placeholder
|
|
subscriberVar <- newTVarIO emptyRMap
|
|
ownSubsVar <- newTVarIO HMap.empty
|
|
ownPostVar <- newTVarIO HSet.empty
|
|
relayInQueue' <- newTQueueIO
|
|
postFetchQueue' <- newTQueueIO
|
|
let
|
|
thisService = PostService {
|
|
serviceConf = conf
|
|
, baseDHT = dht
|
|
, serviceThread = threadVar
|
|
, subscribers = subscriberVar
|
|
, ownSubscriptions = ownSubsVar
|
|
, ownPosts = ownPostVar
|
|
, relayInQueue = relayInQueue'
|
|
, postFetchQueue = postFetchQueue'
|
|
}
|
|
port' = fromIntegral (confServicePort conf)
|
|
warpSettings = Warp.setPort port' . Warp.setHost (fromString . confServiceHost $ conf) $ Warp.defaultSettings
|
|
-- Run 'concurrently_' from another thread to be able to return the
|
|
-- 'PostService'.
|
|
-- Terminating that parent thread will make all child threads terminate as well.
|
|
servThreadID <- forkIO $
|
|
concurrently_
|
|
-- web server
|
|
(Warp.runSettings warpSettings $ postServiceApplication thisService)
|
|
(processIncomingPosts thisService)
|
|
-- update thread ID after fork
|
|
atomically $ writeTVar threadVar servThreadID
|
|
pure thisService
|
|
|
|
getListeningPortFromService = fromIntegral . confServicePort . serviceConf
|
|
|
|
|
|
-- | return a WAI application
|
|
postServiceApplication :: PostService d -> Application
|
|
postServiceApplication serv = serve exposedPostServiceAPI $ postServer serv
|
|
|
|
|
|
-- | needed for guiding type inference
|
|
exposedPostServiceAPI :: Proxy PostServiceAPI
|
|
exposedPostServiceAPI = Proxy
|
|
|
|
-- ========= constants ===========
|
|
|
|
placeholderPost :: Txt.Text
|
|
placeholderPost = Txt.take 5120 . Txt.repeat $ 'O' -- size 5KiB
|
|
|
|
-- ========= HTTP API and handlers =============
|
|
|
|
type PostServiceAPI = "relay" :> "inbox" :> Capture "hashtag" Txt.Text :> ReqBody '[PlainText] Txt.Text :> PutCreated '[PlainText] NoContent
|
|
-- delivery endpoint of newly published posts of the relay's instance
|
|
:<|> "relay" :> "subscribers" :> ReqBody '[PlainText] Txt.Text :> PostNoContent '[PlainText] NoContent
|
|
-- endpoint for delivering the subscriptions and outstanding queue
|
|
:<|> "post" :> Capture "postid" Txt.Text :> Get '[PlainText] Txt.Text
|
|
-- fetch endpoint for posts, full post ID is http://$domain/post/$postid
|
|
:<|> "posts" :> ReqBody '[PlainText] Txt.Text :> Post '[PlainText] Txt.Text
|
|
-- endpoint for fetching multiple posts at once
|
|
:<|> "posts" :> "inbox" :> ReqBody '[PlainText] Txt.Text :> PutCreated '[PlainText] NoContent
|
|
-- delivery endpoint of newly published posts of the relay's instance
|
|
:<|> "tags" :> Capture "hashtag" Txt.Text :> ReqBody '[PlainText] Txt.Text :> PostCreated '[PlainText] Txt.Text
|
|
-- delivery endpoint for posts of $tag at subscribing instance
|
|
:<|> "tags" :> Capture "hashtag" Txt.Text :> "subscribe" :> Header "Origin" Txt.Text :> Get '[PlainText] Integer
|
|
-- endpoint for subscribing the instance specified in
|
|
-- the Origin header to $hashtag.
|
|
-- Returns subscription lease time in seconds.
|
|
:<|> "tags" :> Capture "hashtag" Txt.Text :> "unsubscribe" :> Header "Origin" Txt.Text :> Get '[PlainText] Txt.Text
|
|
-- endpoint for unsubscribing the instance specified in
|
|
-- the Origin header to $hashtag
|
|
|
|
|
|
postServer :: PostService d -> Server PostServiceAPI
|
|
postServer service = relayInbox service
|
|
:<|> subscriptionDelivery service
|
|
:<|> postFetch service
|
|
:<|> postMultiFetch service
|
|
:<|> postInbox service
|
|
:<|> tagDelivery service
|
|
:<|> tagSubscribe service
|
|
:<|> tagUnsubscribe service
|
|
|
|
|
|
relayInbox :: PostService d -> Hashtag -> Txt.Text -> Handler NoContent
|
|
relayInbox serv tag posts = do
|
|
let
|
|
-- skip checking whether the post actually contains the tag, just drop full post
|
|
postIDs = head . Txt.splitOn "," <$> Txt.lines posts
|
|
broadcastChan <- liftIO $ atomically $ getTagBroadcastChannel serv tag
|
|
-- if tag is not in own responsibility, return a 410 Gone
|
|
maybe
|
|
(throwError $ err410 { errBody = "Relay is not responsible for this tag"})
|
|
-- otherwise enqueue posts into broadcast queue of the tag
|
|
(\queue ->
|
|
liftIO $ forM_ postIDs (atomically . writeTChan queue)
|
|
)
|
|
broadcastChan
|
|
pure NoContent
|
|
|
|
subscriptionDelivery :: PostService d -> Txt.Text -> Handler NoContent
|
|
subscriptionDelivery serv subList = do
|
|
let
|
|
tagSubs = Txt.lines subList
|
|
liftIO $ forM_ tagSubs $ processTag (subscribers serv)
|
|
pure NoContent
|
|
-- TODO: check and only accept tags in own (future?) responsibility
|
|
where
|
|
processTag :: TVar RelayTags -> Txt.Text -> IO ()
|
|
processTag subscriberSTM tagData = do
|
|
let
|
|
tag:subText:lease:posts:_ = Txt.splitOn "," tagData
|
|
-- ignore checking of lease time
|
|
leaseTime = fromIntegral (read . Txt.unpack $ lease :: Integer)
|
|
sub = read . Txt.unpack $ subText :: (String, Int)
|
|
postList = Txt.words posts
|
|
enqueueSubscription subscriberSTM (normaliseTag tag) sub postList leaseTime
|
|
|
|
|
|
postFetch :: PostService d -> Txt.Text -> Handler Txt.Text
|
|
postFetch serv postID = do
|
|
postSet <- liftIO . readTVarIO . ownPosts $ serv
|
|
if HSet.member postID postSet
|
|
-- decision: always return the same placeholder post
|
|
then pure placeholderPost
|
|
else throwError $ err404 { errBody = "No post found with this ID" }
|
|
|
|
|
|
postMultiFetch :: PostService d -> Txt.Text -> Handler Txt.Text
|
|
postMultiFetch serv postIDs = do
|
|
let idList = Txt.lines postIDs
|
|
postSet <- liftIO . readTVarIO . ownPosts $ serv
|
|
-- look up existence of all given post IDs, fail if even one is missing
|
|
foldM (\response postID ->
|
|
if HSet.member postID postSet
|
|
then pure $ placeholderPost <> "\n" <> response
|
|
else throwError $ err404 { errBody = "No post found with this ID" }
|
|
) "" idList
|
|
|
|
|
|
postInbox :: PostService d -> Txt.Text -> Handler NoContent
|
|
postInbox serv post = do
|
|
-- extract contained hashtags
|
|
let
|
|
containedTags = fmap (normaliseTag . Txt.tail) . filter ((==) '#' . Txt.head) . Txt.words $ post
|
|
-- generate post ID
|
|
postId <- liftIO $ Txt.pack . show <$> (randomRIO (0, 2^(128::Integer)-1) :: IO Integer)
|
|
-- add ID to own posts
|
|
liftIO . atomically $ modifyTVar' (ownPosts serv) (HSet.insert postId)
|
|
-- enqueue a relay job for each tag
|
|
liftIO $ forM_ (containedTags :: [Txt.Text]) (\tag ->
|
|
atomically $ writeTQueue (relayInQueue serv) (tag, postId, post)
|
|
)
|
|
pure NoContent
|
|
|
|
|
|
tagDelivery :: PostService d -> Txt.Text -> Txt.Text -> Handler Txt.Text
|
|
tagDelivery serv hashtag posts = do
|
|
let postIDs = Txt.lines posts
|
|
subscriptions <- liftIO . readTVarIO . ownSubscriptions $ serv
|
|
if isJust (HMap.lookup (genKeyID . Txt.unpack $ hashtag) subscriptions)
|
|
then -- TODO: increase a counter/ statistics for received posts of this tag
|
|
liftIO $ forM_ postIDs $ atomically . writeTQueue (postFetchQueue serv)
|
|
else -- silently drop posts from unsubscribed tags
|
|
pure ()
|
|
pure $ "Received a postID for tag " <> hashtag
|
|
|
|
tagSubscribe :: PostService d -> Txt.Text -> Maybe Txt.Text -> Handler Integer
|
|
tagSubscribe serv hashtag origin = do
|
|
originURL <- maybe
|
|
(throwError $ err400 { errBody = "Missing Origin header" })
|
|
pure
|
|
origin
|
|
req <- HTTP.parseUrlThrow (Txt.unpack originURL)
|
|
now <- liftIO getPOSIXTime
|
|
let leaseTime = now + confSubscriptionExpiryTime (serviceConf serv)
|
|
-- setup subscription entry
|
|
_ <- liftIO . atomically $ setupSubscriberChannel (subscribers serv) hashtag (BSU.toString $ HTTP.host req, HTTP.port req) leaseTime
|
|
pure $ round leaseTime
|
|
|
|
|
|
tagUnsubscribe :: PostService d -> Txt.Text -> Maybe Txt.Text -> Handler Txt.Text
|
|
tagUnsubscribe serv hashtag origin = do
|
|
originURL <- maybe
|
|
(throwError $ err400 { errBody = "Missing Origin header" })
|
|
pure
|
|
origin
|
|
req <- HTTP.parseUrlThrow (Txt.unpack originURL)
|
|
liftIO . atomically $ deleteSubscription (subscribers serv) hashtag (BSU.toString $ HTTP.host req, HTTP.port req)
|
|
pure "bye bye"
|
|
|
|
-- client/ request functions
|
|
|
|
clientAPI :: Proxy PostServiceAPI
|
|
clientAPI = Proxy
|
|
|
|
|
|
relayInboxClient :<|> subscriptionDeliveryClient :<|> postFetchClient :<|> postMultiFetchClient :<|> postInboxClient :<|> tagDeliveryClient :<|> tagSubscribeClient :<|> tagUnsubscribeClient = client clientAPI
|
|
|
|
|
|
|
|
-- currently this is unused code
|
|
getClients :: String -> Int -> HTTP.Manager -> Client IO PostServiceAPI
|
|
getClients hostname' port' httpMan = hoistClient clientAPI
|
|
(fmap (either (error . show) id)
|
|
. flip runClientM clientEnv
|
|
)
|
|
(client clientAPI)
|
|
where
|
|
clientEnv = mkClientEnv httpMan (BaseUrl Http hostname' port' "")
|
|
|
|
-- ======= data structure manipulations =========
|
|
|
|
-- | Write all pending posts of a subscriber-tag-combination to its queue.
|
|
-- Sets up all necessary data structures if they are still missing.
|
|
enqueueSubscription :: TVar RelayTags -- tag-subscriber map
|
|
-> Hashtag -- hashtag of pending posts
|
|
-> (String, Int) -- subscriber's connection information
|
|
-> [PostID] -- pending posts
|
|
-> POSIXTime -- lease expiry time
|
|
-> IO ()
|
|
enqueueSubscription tagMapSTM tag subscriber posts leaseTime = do
|
|
-- get the tag output queue and, if necessary, create it
|
|
subChan <- atomically $ setupSubscriberChannel tagMapSTM tag subscriber leaseTime
|
|
forM_ posts (atomically . writeTChan subChan)
|
|
|
|
|
|
-- | STM operation to return the outgoing post queue of a tag to a specified subscriber.
|
|
-- If the queue doesn't exist yet, all necessary data structures are set up accordingly.
|
|
setupSubscriberChannel :: TVar RelayTags -> Hashtag -> (String, Int) -> POSIXTime -> STM (TChan PostID)
|
|
setupSubscriberChannel tagMapSTM tag subscriber leaseTime = do
|
|
tagMap <- readTVar tagMapSTM
|
|
case lookupTagSubscriptions tag tagMap of
|
|
Nothing -> do
|
|
-- if no collision/ tag doesn't exist yet, just initialize a
|
|
-- new subscriber map
|
|
broadcastChan <- newBroadcastTChan
|
|
tagOutChan <- dupTChan broadcastChan
|
|
newSubMapSTM <- newTVar $ HMap.singleton subscriber (tagOutChan, leaseTime)
|
|
writeTVar tagMapSTM $ addRMapEntry (genKeyID . Txt.unpack $ tag) (newSubMapSTM, broadcastChan, tag) tagMap
|
|
pure tagOutChan
|
|
Just (foundSubMapSTM, broadcastChan, _) -> do
|
|
-- otherwise use the existing subscriber map
|
|
foundSubMap <- readTVar foundSubMapSTM
|
|
case HMap.lookup subscriber foundSubMap of
|
|
Nothing -> do
|
|
-- for new subscribers, create new output channel
|
|
tagOutChan <- dupTChan broadcastChan
|
|
writeTVar foundSubMapSTM $ HMap.insert subscriber (tagOutChan, leaseTime) foundSubMap
|
|
pure tagOutChan
|
|
-- existing subscriber's channels are just returned
|
|
Just (tagOutChan, _) -> pure tagOutChan
|
|
|
|
|
|
-- | deletes a subscription from the passed subscriber map
|
|
deleteSubscription :: TVar RelayTags -> Hashtag -> (String, Int) -> STM ()
|
|
deleteSubscription tagMapSTM tag subscriber = do
|
|
tagMap <- readTVar tagMapSTM
|
|
case lookupTagSubscriptions tag tagMap of
|
|
-- no subscribers to that tag, just return
|
|
Nothing -> pure ()
|
|
Just (foundSubMapSTM, _, _) -> do
|
|
foundSubMap <- readTVar foundSubMapSTM
|
|
let newSubMap = HMap.delete subscriber foundSubMap
|
|
-- if there are no subscriptions for the tag anymore, remove its
|
|
-- data sttructure altogether
|
|
if HMap.null newSubMap
|
|
then writeTVar tagMapSTM $ deleteRMapEntry (genKeyID . Txt.unpack $ tag) tagMap
|
|
-- otherwise just remove the subscription of that node
|
|
else writeTVar foundSubMapSTM newSubMap
|
|
|
|
|
|
|
|
-- | returns the broadcast channel of a hashtag if there are any subscribers to it
|
|
getTagBroadcastChannel :: PostService d -> Hashtag -> STM (Maybe (TChan PostID))
|
|
getTagBroadcastChannel serv tag = do
|
|
tagMap <- readTVar $ subscribers serv
|
|
case lookupTagSubscriptions tag tagMap of
|
|
Nothing -> pure Nothing
|
|
Just (subscriberSTM, broadcastChan, _) -> do
|
|
subscriberMap <- readTVar subscriberSTM
|
|
if HMap.null subscriberMap
|
|
then pure Nothing
|
|
else pure (Just broadcastChan)
|
|
|
|
|
|
-- | look up the subscription data of a tag
|
|
lookupTagSubscriptions :: Hashtag -> RingMap NodeID a -> Maybe a
|
|
lookupTagSubscriptions tag = rMapLookup (genKeyID . Txt.unpack $ tag)
|
|
|
|
|
|
-- normalise the unicode representation of a string to NFC
|
|
normaliseTag :: Txt.Text -> Txt.Text
|
|
normaliseTag = Txt.fromStrict . normalize NFC . Txt.toStrict
|
|
|
|
-- | define how to convert all showable types to PlainText
|
|
-- No idea what I'm doing with these overlappable instances though ¯\_(ツ)_/¯
|
|
-- TODO: figure out how this overlapping stuff actually works https://downloads.haskell.org/~ghc/latest/docs/html/users_guide/glasgow_exts.html#instance-overlap
|
|
instance {-# OVERLAPPABLE #-} Show a => MimeRender PlainText a where
|
|
mimeRender _ = BSUL.fromString . show
|
|
|
|
instance {-# OVERLAPPABLE #-} Read a => MimeUnrender PlainText a where
|
|
mimeUnrender _ = readEither . BSUL.toString
|
|
|
|
-- ====== worker threads ======
|
|
|
|
-- | process the pending relays of incoming posts from the internal queue:
|
|
-- Look up responsible relay node for given hashtag and forward post to it
|
|
processIncomingPosts :: DHT d => PostService d -> IO ()
|
|
processIncomingPosts serv = forever $ do
|
|
-- blocks until available
|
|
-- TODO: process multiple in parallel
|
|
(tag, pID, pContent) <- atomically . readTQueue $ relayInQueue serv
|
|
lookupRes <- lookupKey (baseDHT serv) (Txt.unpack tag)
|
|
case lookupRes of
|
|
-- no vserver active => wait and retry
|
|
Nothing -> threadDelay $ 10 * 10^6
|
|
Just (responsibleHost, responsiblePort) -> do
|
|
httpMan <- HTTP.newManager HTTP.defaultManagerSettings
|
|
resp <- runClientM (relayInboxClient tag (pID <> "," <> pContent)) (mkClientEnv httpMan (BaseUrl Http responsibleHost (fromIntegral responsiblePort) ""))
|
|
case resp of
|
|
Left err -> putStrLn $ "Error: " <> show err
|
|
Right yay -> putStrLn $ "Yay! " <> show yay
|