summaryrefslogtreecommitdiff
path: root/Que/Server.hs
diff options
context:
space:
mode:
Diffstat (limited to 'Que/Server.hs')
-rw-r--r--Que/Server.hs240
1 files changed, 240 insertions, 0 deletions
diff --git a/Que/Server.hs b/Que/Server.hs
new file mode 100644
index 0000000..841cbfa
--- /dev/null
+++ b/Que/Server.hs
@@ -0,0 +1,240 @@
+{-# LANGUAGE NoImplicitPrelude #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE DeriveGeneric #-}
+
+-- | Interprocess communication
+--
+-- Prior art:
+-- - <https://github.com/jb55/httpipe>
+-- - <https://patchbay.pub>
+-- - <https://github.com/hargettp/courier>
+-- - sorta: <https://ngrok.com/> and <https://localtunnel.github.io/www/>
+--
+-- : exe que-server
+--
+-- : dep async
+-- : dep envy
+-- : dep protolude
+-- : dep scotty
+-- : dep stm
+-- : dep unagi-chan
+-- : dep unordered-containers
+module Que.Server
+ ( main
+ )
+where
+
+import Alpha hiding ( Text
+ , get
+ , gets
+ , modify
+ , poll
+ )
+import qualified Control.Concurrent.Go as Go
+import qualified Control.Concurrent.STM as STM
+import qualified Control.Exception as Exception
+import Control.Monad.Reader ( MonadTrans )
+import qualified Data.ByteString.Builder.Extra as Builder
+import qualified Data.ByteString.Lazy as BSL
+import Data.HashMap.Lazy ( HashMap )
+import qualified Data.HashMap.Lazy as HashMap
+import qualified Data.Text.Encoding as Encoding
+import Data.Text.Lazy ( Text
+ , fromStrict
+ )
+import qualified Data.Text.Lazy.IO as Text
+import qualified Network.HTTP.Types.Status as Http
+import qualified Network.Wai as Wai
+import qualified Network.Wai.Handler.Warp as Warp
+import Network.Wai.Middleware.RequestLogger
+ ( logStdout )
+import qualified System.Envy as Envy
+import qualified System.Exit as Exit
+import qualified Web.Scotty.Trans as Scotty
+
+main :: IO ()
+main = Exception.bracket startup shutdown <| uncurry Warp.run
+ where
+ startup = Envy.decodeEnv >>= \case
+ Left e -> Exit.die e
+ Right c -> do
+ sync <- STM.newTVarIO initialAppState
+ let runActionToIO m = runReaderT (runApp m) sync
+ waiapp <- Scotty.scottyAppT runActionToIO routes
+ putText <| "port:" <> (show <| quePort c)
+ return (quePort c, waiapp)
+ shutdown :: a -> IO a
+ shutdown = pure . identity
+
+newtype App a = App
+ { runApp :: ReaderT (STM.TVar AppState) IO a
+ }
+ deriving (Applicative, Functor, Monad, MonadIO, MonadReader
+ (STM.TVar AppState))
+
+data AppState = AppState
+ { ques :: HashMap Namespace Quebase
+ }
+
+initialAppState :: AppState
+initialAppState = AppState { ques = mempty }
+
+data Config = Config
+ { quePort :: Warp.Port -- ^ QUE_PORT
+ } deriving (Generic, Show)
+
+instance Envy.DefConfig Config where
+ defConfig = Config 3000
+
+instance Envy.FromEnv Config
+
+routes :: Scotty.ScottyT Text App ()
+routes = do
+ Scotty.middleware logStdout
+
+ let quepath = "^\\/([[:alnum:]_]+)\\/([[:alnum:]._/]*)$"
+ let namespace = "^\\/([[:alnum:]_]+)\\/?$" -- matches '/ns' and '/ns/' but not '/ns/path'
+
+ -- | GET /index.html
+ Scotty.get (Scotty.literal "/index.html") <| Scotty.redirect "/_/index"
+ Scotty.get (Scotty.literal "/") <| Scotty.redirect "/_/index"
+
+ -- | GET /_/dash
+ Scotty.get (Scotty.literal "/_/dash") <| do
+ authkey <- fromMaybe "" </ Scotty.header "Authorization"
+ adminkey <- liftIO <| lchomp </ Text.readFile "/run/keys/que-admin"
+ if authkey == adminkey
+ then do
+ d <- app <| gets ques
+ Scotty.json d
+ else do
+ Scotty.status Http.methodNotAllowed405
+ Scotty.text "not allowed"
+
+
+ -- | Namespace management
+ Scotty.matchAny (Scotty.regex namespace) <| do
+ Scotty.status Http.notImplemented501
+ Scotty.text "namespace management coming soon"
+
+ -- | GET que
+ --
+ -- Receive a value from a que. Blocks until a value is received,
+ -- then returns. If 'poll=true', then stream data from the Que to the
+ -- client.
+ Scotty.get (Scotty.regex quepath) <| do
+ (ns, qp) <- extract
+ app . modify <| upsertNamespace ns
+ q <- app <| que ns qp
+ poll <- Scotty.param "poll" !: (pure . const False)
+ guardNs ns ["pub", "_"]
+ case poll of
+ True -> Scotty.stream $ streamQue q
+ _ -> do
+ r <- liftIO <| Go.read q
+ Scotty.html <| fromStrict <| Encoding.decodeUtf8 r
+
+ -- | POST que
+ --
+ -- Put a value on a que. Returns immediately.
+ Scotty.post (Scotty.regex quepath) <| do
+ authkey <- fromMaybe "" </ Scotty.header "Authorization"
+ adminkey <- liftIO <| lchomp </ Text.readFile "/run/keys/que-admin"
+ (ns, qp) <- extract
+ -- Only allow my IP or localhost to publish to '_' namespace
+ when ("_" == ns && authkey /= adminkey)
+ <| Scotty.status Http.methodNotAllowed405
+ >> Scotty.text "not allowed: _ is a reserved namespace"
+ >> Scotty.finish
+ guardNs ns ["pub", "_"]
+ -- passed all auth checks
+ app . modify <| upsertNamespace ns
+ q <- app <| que ns qp
+ qdata <- Scotty.body
+ _ <- liftIO <| Go.write q <| BSL.toStrict qdata
+ return ()
+
+-- | Given `guardNs ns whitelist`, if `ns` is not in the `whitelist`
+-- list, return a 405 error.
+guardNs :: Text -> [Text] -> Scotty.ActionT Text App ()
+guardNs ns whitelist = when (not <| ns `elem` whitelist) <| do
+ Scotty.status Http.methodNotAllowed405
+ Scotty.text
+ <| "not allowed: use 'pub' namespace or signup to protect '"
+ <> ns
+ <> "' at https://que.run"
+ Scotty.finish
+
+-- | recover from a scotty-thrown exception.
+(!:)
+ :: Scotty.ActionT Text App a -- ^ action that might throw
+ -> (Text -> Scotty.ActionT Text App a) -- ^ a function providing a default response instead
+ -> Scotty.ActionT Text App a
+(!:) = Scotty.rescue
+
+-- | Forever write the data from 'Que' to 'Wai.StreamingBody'.
+streamQue :: Que -> Wai.StreamingBody
+streamQue q write _ = Go.mult q >>= loop
+ where
+ loop c =
+ Go.tap c
+ >>= (write . Builder.byteStringInsert)
+ >> (write <| Builder.byteStringInsert "\n")
+ >> loop c
+
+-- | Gets the thing from the Hashmap. Call's 'error' if key doesn't exist.
+grab :: (Eq k, Hashable k) => k -> HashMap k v -> v
+grab = flip (HashMap.!)
+
+-- | Inserts the namespace in 'AppState' if it doesn't exist.
+upsertNamespace :: Namespace -> AppState -> AppState
+upsertNamespace ns as = if HashMap.member ns (ques as)
+ then as
+ else as { ques = HashMap.insert ns mempty (ques as) }
+
+-- | Inserts the que at the proper 'Namespace' and 'Quepath'.
+insertQue :: Namespace -> Quepath -> Que -> AppState -> AppState
+insertQue ns qp q as = as { ques = newQues }
+ where
+ newQues = HashMap.insert ns newQbase (ques as)
+ newQbase = HashMap.insert qp q <| grab ns <| ques as
+
+extract :: Scotty.ActionT Text App (Namespace, Quepath)
+extract = do
+ ns <- Scotty.param "1"
+ path <- Scotty.param "2"
+ return (ns, path)
+
+-- | A synonym for 'lift' in order to be explicit about when we are
+-- operating at the 'App' layer.
+app :: MonadTrans t => App a -> t App a
+app = lift
+
+-- | Get something from the app state
+gets :: (AppState -> b) -> App b
+gets f = ask >>= liftIO . STM.readTVarIO >>= return . f
+
+-- | Apply a function to the app state
+modify :: (AppState -> AppState) -> App ()
+modify f = ask >>= liftIO . atomically . flip STM.modifyTVar' f
+
+type Namespace = Text -- ^ housing for a set of que paths
+type Que = Go.Channel Message -- ^ a que is just a channel of bytes
+type Quepath = Text -- ^ any path can serve as an identifier for a que
+type Message = ByteString -- ^ any opaque data
+type Quebase = HashMap Quepath Que -- ^ a collection of ques
+
+-- | Lookup or create a que
+que :: Namespace -> Quepath -> App Que
+que ns qp = do
+ _ques <- gets ques
+ let qbase = grab ns _ques
+ queExists = HashMap.member qp qbase
+ if queExists
+ then return <| grab qp qbase
+ else do
+ c <- liftIO <| Go.chan 1
+ modify (insertQue ns qp c)
+ gets ques /> grab ns /> grab qp
generated by cgit v1.2.3 (git 2.25.1) at 2024-12-27 05:13:50 +0000