{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE ScopedTypeVariables #-}
-- | AUTH-8: a Plexus service in miniature, speaking the wire synapse speaks
-- (JSON-RPC over a WebSocket; @_info@, @<backend>.call@, subscription
-- notifications ending in @done@), answering AUTH-2..5's shapes as the
-- tickets write them. It listens on 127.0.0.1, a port the OS picks.
--
-- A 'Scenario' says which service it is: today's (no view, no auth activation),
-- an anonymous caller's, or a signed-in caller's. The person flow's
-- browser is played by the service itself: answering @auth.login {flow:
-- {browser: {redirect}}}@ it also, a moment later, sends the GET the
-- provider's redirect would — to @redirect?code=…&state=…@.
module MockService
( Scenario(..)
, ViewMode(..)
, todaysService
, withMockService
, withEphemeralService
, withMockServiceOn
, serveMockService
, mockCode
, mockState
, catalogService
, schemaCalls
-- * SYNAPSE-28: what a run cost the service
, connectionsAccepted
, infoCalls
, unsubscribes
, renamedGet
-- * SYNAPSE-CC-8: a service serving schemas a test gives it
, schemaOverride
-- * SYNAPSE-40: a dial that takes its time, and what is left open
, upgradeDelay
, connectionsOpen
-- * SYNAPSE-26: what a closed stdout cost the service
, callsMade
, floodsEnded
, dropNext
-- * RETURNS-36: a service on the published 0.7.0 line
, served070
-- * RETURNS-40: a service that writes its own IR
, irAnswer
, irAsked
-- * BIDIR-18: methods with an input half
, InputPlan(..)
, defaultInputPlan
, inputPlan
, inputSchemas
, Received(..)
, inputReceived
, inputViolations
, resetInput
) where
import Control.Concurrent (forkIO, threadDelay)
import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, tryPutMVar, tryTakeMVar)
import System.Timeout (timeout)
import Control.Exception (SomeException, bracket, finally, try)
import Control.Monad (forever, void)
import Data.Aeson
import qualified Data.Aeson.Key as K
import qualified Data.Aeson.KeyMap as KM
import qualified Data.ByteString as BS
import qualified Data.ByteString.Char8 as BS8
import qualified Data.ByteString.Lazy as LBS
import Data.Bits (shiftL, (.|.))
import qualified Data.Map.Strict as Map
import Data.Word (Word64)
import Data.IORef
import Data.Text (Text)
import qualified Data.Text as T
import qualified Data.Text.Encoding as TE
import qualified Network.Socket as N
import qualified Network.Socket.ByteString as NB
import qualified Network.WebSockets as WS
import qualified Data.Vector as V
import System.IO (hFlush, stdout)
import System.IO.Unsafe (unsafePerformIO)
import Plexus07 (withContract)
data ViewMode = NoView | AnonymousView | CallerView
deriving (Eq, Show)
data Scenario = Scenario
{ scView :: ViewMode
, scAuth :: Bool -- ^ has the auth activation (auth.login, auth.whoami, …)
, scSessionLogin :: Bool -- ^ has IDENTITY-42's session.login (today's menger)
, scBrowserState :: Text -- ^ the state the browser comes back with (mockState: the right one)
, scCatalog :: Bool -- ^ SYNAPSE-17: serve menger's tickets and deployments as they were (test/fixtures)
} deriving (Eq, Show)
todaysService :: Scenario
todaysService = Scenario NoView False True mockState False
-- | SYNAPSE-17: a service with a catalog to search — menger's tickets (17 of
-- its methods) and deployments (3), as its schema had them on 2026-10-15;
-- in a caller's view, tickets.delete (admin only) is not in it.
catalogService :: ViewMode -> Scenario
catalogService v = Scenario v False False mockState True
-- | How many schema calls the mock services have answered (this process).
{-# NOINLINE schemaCalls #-}
schemaCalls :: IORef Int
schemaCalls = unsafePerformIO (newIORef 0)
-- | SYNAPSE-28: WebSocket connections the mock services have accepted (this process).
{-# NOINLINE connectionsAccepted #-}
connectionsAccepted :: IORef Int
connectionsAccepted = unsafePerformIO (newIORef 0)
-- | SYNAPSE-28: _info probes answered.
{-# NOINLINE infoCalls #-}
infoCalls :: IORef Int
infoCalls = unsafePerformIO (newIORef 0)
-- | SYNAPSE-28: @menger.call_unsub@ requests, by subscription.
{-# NOINLINE unsubscribes #-}
unsubscribes :: IORef [Value]
unsubscribes = unsafePerformIO (newIORef [])
-- | SYNAPSE-28: the service renamed @tickets.get@ to @tickets.fetch@ (its schema
-- says @fetch@, and @get@ is not found).
{-# NOINLINE renamedGet #-}
renamedGet :: IORef Bool
renamedGet = unsafePerformIO (newIORef False)
-- | SYNAPSE-CC-8: when set, @<ns>.schema@ (@menger.schema@ for the root,
-- keyed @""@) answers these schemas, by dotted path, before anything else.
{-# NOINLINE schemaOverride #-}
schemaOverride :: IORef (Maybe [(Text, Value)])
schemaOverride = unsafePerformIO (newIORef Nothing)
-- | SYNAPSE-40: microseconds each service waits before it answers a
-- WebSocket upgrade (a dial under way that long).
{-# NOINLINE upgradeDelay #-}
upgradeDelay :: IORef Int
upgradeDelay = unsafePerformIO (newIORef 0)
-- | SYNAPSE-40: WebSocket connections open at the services now.
{-# NOINLINE connectionsOpen #-}
connectionsOpen :: IORef Int
connectionsOpen = unsafePerformIO (newIORef 0)
-- | SYNAPSE-26: the method of every @menger.call@ the services took, in order.
{-# NOINLINE callsMade #-}
callsMade :: IORef [Text]
callsMade = unsafePerformIO (newIORef [])
-- | SYNAPSE-26: @tickets.flood@ streams that ended — unsubscribed, or their
-- connection gone.
{-# NOINLINE floodsEnded #-}
floodsEnded :: IORef Int
floodsEnded = unsafePerformIO (newIORef 0)
-- | SYNAPSE-26: methods whose next call is taken and then answered by
-- closing the connection (each name once).
{-# NOINLINE dropNext #-}
dropNext :: IORef [Text]
dropNext = unsafePerformIO (newIORef [])
-- | RETURNS-36: when set, the services answer as one on the published 0.7.0
-- line does — every schema as written (a method's @streaming@, no
-- @cardinality@), every error frame spec/05's fields alone (@message@,
-- @code@, @error_kind@ and a refusal's fields, no typed @error@).
{-# NOINLINE served070 #-}
served070 :: IORef Bool
served070 = unsafePerformIO (newIORef False)
-- | RETURNS-40: when set, the service writes its own IR: @menger.ir@
-- answers this (@{"ok": IR}@ or @{"err": …}@) as plexus 0.8's does. Unset,
-- it has no @ir@ — every 0.7.0 service — and the call fails as an unknown
-- method does.
{-# NOINLINE irAnswer #-}
irAnswer :: IORef (Maybe Value)
irAnswer = unsafePerformIO (newIORef Nothing)
-- | RETURNS-40: the params of every @menger.ir@ the services took, in order.
{-# NOINLINE irAsked #-}
irAsked :: IORef [Value]
irAsked = unsafePerformIO (newIORef [])
-- | BIDIR-18: how the services answer a call's input half.
data InputPlan = InputPlan
{ planWindow :: Int
-- ^ the first credit, and the window: past half of it consumed, an
-- @ack@ and a @credit@ for what was consumed
, planStingy :: Bool
-- ^ grant the first window and nothing more, never ack
, planRefuse :: Maybe (Int, Text)
-- ^ at this many frames, refuse with this @error_kind@ (naming the frame)
, planClose :: Maybe (Int, Text)
-- ^ at this many frames, @input_closed {reason}@, then an error item
, planDropAfter :: Maybe Int
-- ^ at this many frames, close the socket
} deriving (Eq, Show)
defaultInputPlan :: InputPlan
defaultInputPlan = InputPlan 64 False Nothing Nothing Nothing
{-# NOINLINE inputPlan #-}
inputPlan :: IORef InputPlan
inputPlan = unsafePerformIO (newIORef defaultInputPlan)
-- | One input frame as a service took it.
data Received
= ReceivedJson Word64 [Value] -- ^ @menger.input {seq, items}@
| ReceivedBinary Word64 Word64 BS.ByteString -- ^ a binary frame: subscription, seq, payload
| ReceivedEnd Word64 -- ^ @menger.end {seq}@
deriving (Eq, Show)
-- | BIDIR-18: every input frame the services took, in order.
{-# NOINLINE inputReceived #-}
inputReceived :: IORef [Received]
inputReceived = unsafePerformIO (newIORef [])
-- | BIDIR-18: what a client did that spec/10 forbids (past credit, out of
-- sequence, a malformed binary frame), each said.
{-# NOINLINE inputViolations #-}
inputViolations :: IORef [Text]
inputViolations = unsafePerformIO (newIORef [])
resetInput :: InputPlan -> IO ()
resetInput plan = do
writeIORef inputPlan plan
writeIORef inputReceived []
writeIORef inputViolations []
-- | BIDIR-18: a root with a @stream@ activation beside @tickets@, whose
-- methods take a stream in — for 'schemaOverride'.
inputSchemas :: [(Text, Value)]
inputSchemas =
[ ("", object
[ "namespace" .= ("menger" :: Text), "version" .= ("1.0.0" :: Text), "description" .= ("Central routing" :: Text)
, "hash" .= ("r-input" :: Text)
, "children" .=
[ object ["namespace" .= ("tickets" :: Text), "description" .= ("Plans as data" :: Text), "hash" .= ("t0" :: Text)]
, object ["namespace" .= ("stream" :: Text), "description" .= ("Streams in" :: Text), "hash" .= ("s0" :: Text)] ] ])
, ("stream", object
[ "namespace" .= ("stream" :: Text), "version" .= ("0.1.0" :: Text), "hash" .= ("s0" :: Text)
, "description" .= ("Streams in" :: Text)
, "methods" .=
[ m "lines" "A log, a line an item" (object ["type" .= ("string" :: Text)]) Nothing
, m "ingest" "Items in, an object each" item Nothing
, m "audio" "Raw bytes in" bytesSchema Nothing
, m "tagged" "Lines in, beside a parameter named input" (object ["type" .= ("string" :: Text)])
(Just (object ["type" .= ("object" :: Text), "required" .= (["input"] :: [Text]), "properties" .= object
["input" .= object ["type" .= ("string" :: Text), "description" .= ("A tag" :: Text)]]]))
] ]) ]
where
m :: Text -> Text -> Value -> Maybe Value -> Value
m n d inp ps = object $
[ "name" .= (n :: Text), "description" .= (d :: Text), "hash" .= (n <> "0")
, "returns" .= object [], "input" .= inp ] ++ [ "params" .= p | Just p <- [ps] ]
item = object
[ "type" .= ("object" :: Text), "title" .= ("Item" :: Text), "required" .= (["at", "pad"] :: [Text])
, "properties" .= object ["at" .= object ["type" .= ("integer" :: Text)], "pad" .= object ["type" .= ("string" :: Text)]] ]
bytesSchema = object
[ "type" .= ("string" :: Text), "contentEncoding" .= ("base64" :: Text)
, "contentMediaType" .= ("application/octet-stream" :: Text), "x-plexus-binary" .= True ]
-- | One call's input half at the service.
data InputCall = InputCall
{ icEmit :: Value -> IO ()
, icMeta :: Value
, icMethod :: Text
, icNext :: IORef Word64 -- ^ the next frame expected
, icItems :: IORef Int -- ^ items taken
, icGranted :: IORef Int -- ^ items granted in all
, icUngiven :: IORef Int -- ^ items consumed and not yet granted back
, icTag :: Maybe Text -- ^ the call's own param named input, if it had one
, icOver :: IORef Bool -- ^ the half is closed (refused, closed)
}
mockCode, mockState :: Text
mockCode = "c_19ab"
mockState = "s_7f2c"
-- | Run an action against a service on a fresh port; the port is passed in.
withMockService :: Scenario -> (Int -> IO a) -> IO a
withMockService sc act = bracket open (N.close . fst) (\(_, port) -> act port)
where
open = do
sock <- N.socket N.AF_INET N.Stream N.defaultProtocol
N.setSocketOption sock N.ReuseAddr 1
N.bind sock (N.SockAddrInet 0 (N.tupleToHostAddress (127, 0, 0, 1)))
N.listen sock 16
port <- fromIntegral <$> N.socketPort sock
void $ forkIO $ acceptLoop (const (pure ())) sc sock
pure (sock, port)
-- | SYNAPSE-40: today's service on a port already chosen (one a stopped
-- service had: a server come back).
withMockServiceOn :: Int -> IO a -> IO a
withMockServiceOn port act = bracket open N.close (const act)
where
open = do
sock <- N.socket N.AF_INET N.Stream N.defaultProtocol
N.setSocketOption sock N.ReuseAddr 1
N.bind sock (N.SockAddrInet (fromIntegral port) (N.tupleToHostAddress (127, 0, 0, 1)))
N.listen sock 16
void $ forkIO $ acceptLoop (const (pure ())) todaysService sock
pure sock
-- | SYNAPSE-40: a service that stops as a process does. When the action
-- ends, its listener closes and so does every connection it accepted (a
-- FIN, as the kernel sends for a process that exits), so a client still
-- holding one sees its server go.
withEphemeralService :: Scenario -> (Int -> IO a) -> IO a
withEphemeralService sc act = do
accepted <- newIORef []
bracket (open accepted) (stop accepted) (\(_, port) -> act port)
where
open accepted = do
sock <- N.socket N.AF_INET N.Stream N.defaultProtocol
N.setSocketOption sock N.ReuseAddr 1
N.bind sock (N.SockAddrInet 0 (N.tupleToHostAddress (127, 0, 0, 1)))
N.listen sock 16
port <- fromIntegral <$> N.socketPort sock
void $ forkIO $ acceptLoop (\c -> atomicModifyIORef' accepted (\cs -> (c : cs, ()))) sc sock
pure (sock, port)
stop accepted (sock, _) = do
N.close sock
cs <- readIORef accepted
mapM_ (\c -> void (try (N.shutdown c N.ShutdownBoth) :: IO (Either SomeException ()))) cs
-- | Serve until killed, printing the port (for transcripts by hand).
serveMockService :: Scenario -> IO ()
serveMockService sc = withMockService sc $ \port -> do
putStrLn (show port)
hFlush stdout
forever (threadDelay 1000000)
acceptLoop :: (N.Socket -> IO ()) -> Scenario -> N.Socket -> IO ()
acceptLoop onAccept sc sock = do
r <- try (N.accept sock)
case r of
Left (_ :: SomeException) -> pure () -- the service was closed
Right (conn, _) -> do
onAccept conn
void $ forkIO $ do
_ <- try (session conn) :: IO (Either SomeException ())
N.close conn
acceptLoop onAccept sc sock
where
session conn = do
pending <- WS.makePendingConnection conn WS.defaultConnectionOptions
-- the credential the upgrade carried, if any
let bearer = lookup "Authorization" (WS.requestHeaders (WS.pendingRequest pending))
readIORef upgradeDelay >>= \d -> if d > 0 then threadDelay d else pure ()
c <- WS.acceptRequest pending
atomicModifyIORef' connectionsAccepted (\n -> (n + 1, ()))
atomicModifyIORef' connectionsOpen (\n -> (n + 1, ()))
next <- newIORef (100 :: Int)
inputs <- newIORef Map.empty
flip finally (atomicModifyIORef' connectionsOpen (\n -> (n - 1, ()))) $ forever $ do
dm <- WS.receiveDataMessage c
case dm of
WS.Binary bytes -> binaryInput inputs (LBS.toStrict bytes)
WS.Text msg _ -> case decode msg of
Just (Object o)
| str "method" o `elem` ["menger.input", "menger.end"] -> jsonInput inputs o
| otherwise -> handle sc c next inputs (isBearer bearer) o
_ -> pure ()
isBearer = maybe False (BS8.isPrefixOf "Bearer ")
handle :: Scenario -> WS.Connection -> IORef Int -> IORef (Map.Map Int InputCall) -> Bool -> Object -> IO ()
handle sc c next inputs signedIn o = do
let rid = KM.lookup "id" o
method = str "method" o
params = case KM.lookup "params" o of { Just (Object p) -> p; _ -> KM.empty }
-- SYNAPSE-28: an unsubscribe is answered true, and noted
if method == "menger.call_unsub"
then do
atomicModifyIORef' unsubscribes (\us -> (maybe Null id (KM.lookup "params" o) : us, ()))
send (object ["jsonrpc" .= ("2.0" :: Text), "id" .= rid, "result" .= True])
else handleCall sc c next inputs signedIn o rid method params
where
send :: Value -> IO ()
send = WS.sendTextData c . encode
handleCall :: Scenario -> WS.Connection -> IORef Int -> IORef (Map.Map Int InputCall) -> Bool -> Object -> Maybe Value -> Text -> Object -> IO ()
handleCall sc c next inputs signedIn _o rid method params = do
sub <- atomicModifyIORef' next (\n -> (n + 1, n))
send (object ["jsonrpc" .= ("2.0" :: Text), "id" .= rid, "result" .= sub])
let emit item = send (object
[ "jsonrpc" .= ("2.0" :: Text), "method" .= ("subscription" :: Text)
, "params" .= object ["subscription" .= sub, "result" .= item] ])
meta = object ["provenance" .= (["menger"] :: [Text]), "plexus_hash" .= ("h0" :: Text), "timestamp" .= (0 :: Int)]
-- RETURNS-36: a schema as plexus 0.7 states it (its methods' cardinality and returns)
dat :: Text -> Value -> IO ()
dat ct v = do
as070 <- readIORef served070
emit (object ["type" .= ("data" :: Text), "metadata" .= meta, "content_type" .= ct
, "content" .= (if ".schema" `T.isSuffixOf` ct && not as070 then withContract v else v)])
-- RETURNS-36: plexus 0.7's error frame — the typed RemoteError the
-- message names, and AUTH-2's refusal fields beside it
err :: Text -> [(K.Key, Value)] -> IO ()
err m extra = do
as070 <- readIORef served070
-- as plexus-core writes it: message, code and spec/05's fields
-- always; 0.8's typed error beside them (the 0.7.0 line has none)
emit (Object (KM.fromList (
[("type", "error"), ("metadata", meta), ("message", String m), ("code", String (code070 m extra)), ("recoverable", Bool False)]
<> [ ("error", remoteError m extra) | not as070 ] <> extra)))
done = emit (object ["type" .= ("done" :: Text), "metadata" .= meta])
case method of
"_info" -> do
atomicModifyIORef' infoCalls (\n -> (n + 1, ()))
dat "_info" (infoFor sc)
done
"menger.call" -> do
let inner = str "method" params
args = case KM.lookup "params" params of { Just (Object a) -> a; _ -> KM.empty }
renamed <- readIORef renamedGet
atomicModifyIORef' callsMade (\cs -> (cs ++ [inner], ()))
dropIt <- atomicModifyIORef' dropNext (\ds -> if inner `elem` ds then (filter (/= inner) ds, True) else (ds, False))
if dropIt then ioError (userError "dropped") else pure ()
case inner of
-- SYNAPSE-26: a write answered in two parts, a moment apart, so a
-- reader that closes after the first has closed before the second
"tickets.amend" -> void $ forkIO $ do
dat "tickets.amend" (object ["type" .= ("amended" :: Text), "id" .= str "id" args, "part" .= (1 :: Int)])
threadDelay 300000
dat "tickets.amend" (object ["type" .= ("amended" :: Text), "id" .= str "id" args, "part" .= (2 :: Int)])
done
-- SYNAPSE-26: a stream that ticks every 100 ms until it is
-- unsubscribed or its connection goes
"tickets.flood" -> void $ forkIO $ do
u0 <- length <$> readIORef unsubscribes
let tick (n :: Int) = do
u <- length <$> readIORef unsubscribes
if u > u0 then pure () else do
ok <- try (dat "tickets.flood" (object ["type" .= ("tick" :: Text), "n" .= n]))
case ok of
Left (_ :: SomeException) -> pure ()
Right () -> threadDelay 100000 >> tick (n + 1)
tick 0
atomicModifyIORef' floodsEnded (\n -> (n + 1, ()))
-- waits for its answer off the connection's loop, as a real service
-- runs calls side by side (SYNAPSE-28: the answer may come on this
-- very connection)
"tickets.ask" -> void $ forkIO (askBack emit meta dat err >> done)
-- SYNAPSE-28: a stream that never ends by itself (until unsubscribed)
"tickets.watch" -> dat "tickets.watch" (object ["type" .= ("watching" :: Text)])
-- BIDIR-18: a method with an input half: credit at once, then the
-- frames as they come (jsonInput, binaryInput)
_ | inner `elem` ["stream.lines", "stream.ingest", "stream.audio", "stream.tagged"] -> do
plan <- readIORef inputPlan
ic <- InputCall emit meta inner <$> newIORef 0 <*> newIORef 0 <*> newIORef (planWindow plan)
<*> newIORef 0 <*> pure (case KM.lookup "input" args of { Just (String t) -> Just t; _ -> Nothing })
<*> newIORef False
atomicModifyIORef' inputs (\m -> (Map.insert sub ic m, ()))
emit (object ["type" .= ("credit" :: Text), "n" .= planWindow plan])
"menger.ir" -> do
atomicModifyIORef' irAsked (\as -> (as ++ [maybe (Object KM.empty) id (KM.lookup "params" params)], ()))
answer <- readIORef irAnswer
case answer of
Just v -> dat "menger.ir" v >> done
Nothing -> err "Method not found: menger.ir" [] >> done
"tickets.get" | renamed -> err "Method not found: tickets.get" [] >> done
"tickets.fetch" | renamed -> callMethod sc signedIn "tickets.get" args dat err >> done
"tickets.schema" | renamed, scView sc /= AnonymousView, not (scCatalog sc) ->
dat "tickets.schema" (renameGet (ticketsSchema sc)) >> done
_ -> callMethod sc signedIn inner args dat err >> done
-- SCAFFOLD-21: a bidirectional answer, on a connection of its own
"menger.respond" -> do
void (tryPutMVar responses (KM.lookup "response_data" params))
done
_ -> err ("Invalid method format: " <> method) [] >> done
where
send :: Value -> IO ()
send = WS.sendTextData c . encode
-- | BIDIR-18: a JSON input frame or end, for the call it names.
jsonInput :: IORef (Map.Map Int InputCall) -> Object -> IO ()
jsonInput inputs o = do
let params = case KM.lookup "params" o of { Just (Object p) -> p; _ -> KM.empty }
num k = case KM.lookup k params of { Just (Number n) -> Just (round n); _ -> Nothing }
case (num "subscription", num "seq") of
(Just sub, Just sq) -> do
calls <- readIORef inputs
case Map.lookup sub calls of
Nothing -> violation ("a frame for no call: " <> T.pack (show sub))
Just ic
| str "method" o == "menger.end" -> do
atomicModifyIORef' inputReceived (\rs -> (rs ++ [ReceivedEnd (fromIntegral sq)], ()))
expected <- readIORef (icNext ic)
if fromIntegral sq /= expected then violation ("end seq " <> T.pack (show sq) <> ", frames " <> T.pack (show expected)) else pure ()
finish ic
| otherwise -> do
let items = case KM.lookup "items" params of { Just (Array a) -> V.toList a; _ -> [] }
atomicModifyIORef' inputReceived (\rs -> (rs ++ [ReceivedJson (fromIntegral sq) items], ()))
taken ic (fromIntegral sq) (length items)
_ -> violation "an input frame without subscription or seq"
-- | BIDIR-18: a binary input frame: @[0x01][subscription u64][seq u64][payload]@.
binaryInput :: IORef (Map.Map Int InputCall) -> BS.ByteString -> IO ()
binaryInput inputs bytes
| BS.length bytes < 17 || BS.head bytes /= 0x01 = violation "a malformed binary frame"
| otherwise = do
let word b = foldl (\acc x -> (acc `shiftL` 8) .|. fromIntegral x) (0 :: Word64) (BS.unpack b)
sub = word (BS.take 8 (BS.drop 1 bytes))
sq = word (BS.take 8 (BS.drop 9 bytes))
payload = BS.drop 17 bytes
atomicModifyIORef' inputReceived (\rs -> (rs ++ [ReceivedBinary sub sq payload], ()))
calls <- readIORef inputs
case Map.lookup (fromIntegral sub) calls of
Nothing -> violation ("a binary frame for no call: " <> T.pack (show sub))
Just ic -> taken ic sq 1
violation :: Text -> IO ()
violation v = atomicModifyIORef' inputViolations (\vs -> (vs ++ [v], ()))
-- | Frame @sq@ of @n@ items taken: checked against the sequence and the
-- credit, then acked and credited as the plan says.
taken :: InputCall -> Word64 -> Int -> IO ()
taken ic sq n = do
over <- readIORef (icOver ic)
if over then pure () else do
plan <- readIORef inputPlan
expected <- readIORef (icNext ic)
if sq /= expected then violation ("out of sequence: " <> T.pack (show sq) <> ", expected " <> T.pack (show expected)) else pure ()
writeIORef (icNext ic) (sq + 1)
total <- atomicModifyIORef' (icItems ic) (\k -> (k + n, k + n))
granted <- readIORef (icGranted ic)
if total > granted then violation ("past credit: " <> T.pack (show total) <> " of " <> T.pack (show granted)) else pure ()
let frames = fromIntegral (sq + 1)
errorItem kind = icEmit ic (object
[ "type" .= ("error" :: Text), "metadata" .= icMeta ic, "message" .= ("input refused: " <> kind)
, "code" .= ("-32602" :: Text), "error_kind" .= kind, "recoverable" .= False, "seq" .= sq ])
case (planDropAfter plan, planRefuse plan, planClose plan) of
(Just k, _, _) | frames >= k -> ioError (userError "dropped mid-input")
(_, Just (k, kind), _) | frames >= k -> do
writeIORef (icOver ic) True
icEmit ic (object ["type" .= ("input_closed" :: Text), "reason" .= kind])
errorItem kind
(_, _, Just (k, reason)) | frames >= k -> do
writeIORef (icOver ic) True
icEmit ic (object ["type" .= ("input_closed" :: Text), "reason" .= reason, "ack" .= sq])
icEmit ic (object
[ "type" .= ("error" :: Text), "metadata" .= icMeta ic, "message" .= ("the call was asked again: " <> reason)
, "code" .= ("-32001" :: Text), "error_kind" .= ("unauthenticated" :: Text), "recoverable" .= False ])
_ | planStingy plan -> pure ()
| otherwise -> do
ungiven <- atomicModifyIORef' (icUngiven ic) (\u -> (u + n, u + n))
if ungiven * 2 >= planWindow plan
then do
writeIORef (icUngiven ic) 0
modifyIORef' (icGranted ic) (+ ungiven)
icEmit ic (object ["type" .= ("ack" :: Text), "seq" .= sq])
icEmit ic (object ["type" .= ("credit" :: Text), "n" .= ungiven])
else pure ()
-- | The input ended: the last ack, what was taken, done.
finish :: InputCall -> IO ()
finish ic = do
n <- readIORef (icItems ic)
next <- readIORef (icNext ic)
if next > 0 then icEmit ic (object ["type" .= ("ack" :: Text), "seq" .= (next - 1)]) else pure ()
icEmit ic (object
[ "type" .= ("data" :: Text), "metadata" .= icMeta ic, "content_type" .= icMethod ic
, "content" .= object (["type" .= ("taken" :: Text), "items" .= n, "frames" .= next] ++ [ "tag" .= t | Just t <- [icTag ic] ]) ])
icEmit ic (object ["type" .= ("done" :: Text), "metadata" .= icMeta ic])
-- | RETURNS-36: the plexus 0.7 'RemoteError' an error the mock answers is:
-- a refusal (AUTH-2's fields) a missing grant, a "Method not found: a.m" or
-- "Activation not found: a" one of those, anything else an execution failure.
remoteError :: Text -> [(K.Key, Value)] -> Value
remoteError m extra
| lookup "error_kind" extra == Just (String "forbidden") = object
[ "kind" .= ("forbidden" :: Text), "reason" .= ("missing_grant" :: Text)
, "grant" .= field "grant", "scope" .= field "scope"
, "roles" .= maybe (toJSON ([] :: [Text])) id (lookup "roles" extra)
, "held_on" .= maybe (toJSON ([] :: [Text])) id (lookup "held_on" extra) ]
| Just rest <- T.stripPrefix "Method not found: " m
, (a, dm) <- T.breakOnEnd "." rest, not (T.null a) = object
[ "kind" .= ("method_not_found" :: Text), "activation" .= T.dropEnd 1 a, "method" .= dm ]
| Just a <- T.stripPrefix "Activation not found: " m = object
[ "kind" .= ("activation_not_found" :: Text), "activation" .= a ]
| otherwise = object [ "kind" .= ("execution" :: Text), "message" .= m ]
where
field k = maybe (String "") id (lookup k extra)
-- | The JSON-RPC code a service puts on an error frame (spec/05).
code070 :: Text -> [(K.Key, Value)] -> Text
code070 m extra
| lookup "error_kind" extra == Just (String "forbidden") = "-32003"
| "not found" `T.isInfixOf` m = "-32601"
| otherwise = "-32000"
-- | SCAFFOLD-21: the answers to 'askBack's requests (@menger.respond@'s
-- @response_data@), as they arrive on any connection.
{-# NOINLINE responses #-}
responses :: MVar (Maybe Value)
responses = unsafePerformIO newEmptyMVar
-- | A bidirectional method (@tickets.ask@): it asks the caller to confirm
-- (a @request@ item), waits up to 5 s for @menger.respond@, and answers
-- with what came back.
askBack :: (Value -> IO ()) -> Value -> (Text -> Value -> IO ()) -> (Text -> [(K.Key, Value)] -> IO ()) -> IO ()
askBack emit meta dat err = do
_ <- tryTakeMVar responses
emit (object
[ "type" .= ("request" :: Text), "metadata" .= meta, "request_id" .= ("req-1" :: Text)
, "request" .= object ["type" .= ("confirm" :: Text), "message" .= ("go on?" :: Text)] ])
got <- timeout 5000000 (takeMVar responses)
case got of
Just (Just r) -> dat "tickets.ask" (object ["type" .= ("answered" :: Text), "response" .= r])
_ -> err "no response to req-1" []
infoFor :: Scenario -> Value
infoFor sc = case scView sc of
NoView -> object ["backend" .= ("menger" :: Text), "auth_capabilities" .= object ["mechanisms" .= [object ["kind" .= ("anonymous" :: Text)]]]]
AnonymousView -> object
[ "backend" .= ("menger" :: Text), "view" .= ("anonymous" :: Text)
, "auth" .= object
[ "methods" .= (["auth.providers", "auth.login", "auth.signup"] :: [Text])
, "resource_metadata" .= ("http://127.0.0.1:4456/.well-known/oauth-protected-resource" :: Text) ] ]
CallerView -> object ["backend" .= ("menger" :: Text), "view" .= ("caller" :: Text), "view_hash" .= ("v-9a77" :: Text)]
callMethod :: Scenario -> Bool -> Text -> Object
-> (Text -> Value -> IO ()) -> (Text -> [(K.Key, Value)] -> IO ()) -> IO ()
callMethod sc signedIn inner args dat err = do
if ".schema" `T.isSuffixOf` inner then atomicModifyIORef' schemaCalls (\n -> (n + 1, ())) else pure ()
override <- readIORef schemaOverride
let ns = if inner == "menger.schema" then "" else T.dropEnd (T.length ".schema") inner
case override >>= lookup ns of
Just v | ".schema" `T.isSuffixOf` inner -> dat inner v
_ -> callMethod' sc signedIn inner args dat err
callMethod' :: Scenario -> Bool -> Text -> Object
-> (Text -> Value -> IO ()) -> (Text -> [(K.Key, Value)] -> IO ()) -> IO ()
callMethod' sc _signedIn inner args dat err =
case inner of
"menger.schema" -> do
-- RETURNS-40: a service that writes its IR lists `ir` among its own methods
answer <- readIORef irAnswer
dat "menger.schema" (maybe id (const withIr) answer (rootSchema sc))
"tickets.schema" | scCatalog sc -> do
Just (Object o) <- decodeFileStrict "test/fixtures/find-tickets.json"
dat "tickets.schema" (Object (catalogView sc o))
"deployments.schema" | scCatalog sc -> do
Just v <- decodeFileStrict "test/fixtures/find-deployments.json"
dat "deployments.schema" (v :: Value)
"auth.schema" | scAuth sc -> dat "auth.schema" authSchema
"tickets.schema" | scView sc /= AnonymousView -> dat "tickets.schema" (ticketsSchema sc)
"auth.whoami" | scAuth sc -> dat "auth.whoami" (object
[ "type" .= ("whoami" :: Text), "principal" .= ("key:ac79e0f3b1c24d5e6f708192a3b4c5d6e7f8091a2b3c4d5e6f708192a3b4c5d6" :: Text)
, "name" .= ("worker-1" :: Text), "agent" .= ("worker-1" :: Text)
, "memberships" .= [object ["role" .= ("editor" :: Text), "scope" .= ("MENGER" :: Text)]]
, "ceiling" .= (["editor@MENGER-12"] :: [Text]) ])
"tickets.get" -> dat "tickets.get" (object ["type" .= ("ticket" :: Text), "id" .= ("MENGER-12" :: Text)])
"tickets.set_status"
| scView sc == CallerView -> err "needs tickets.set_status on IDENTITY"
[ ("error_kind", "forbidden"), ("grant", "tickets.set_status"), ("scope", "IDENTITY")
, ("roles", toJSON (["editor", "admin"] :: [Text])), ("held_on", toJSON (["MENGER"] :: [Text])) ]
| otherwise -> dat "tickets.set_status" (object ["type" .= ("status_changed" :: Text)])
"auth.login" | scAuth sc -> case KM.lookup "flow" args of
Just (Object f)
| Just (Object b) <- KM.lookup "browser" f -> do
let redirect = str "redirect" b
dat "auth.login" (object
[ "type" .= ("login_url" :: Text)
, "url" .= ("http://127.0.0.1:4456/login?cli=1&state=" <> mockState)
, "state" .= mockState ])
-- the provider's redirect, a moment later
void $ forkIO $ threadDelay 300000 >> browserReturns redirect (scBrowserState sc)
| Just (Object cd) <- KM.lookup "code" f ->
if str "code" cd == mockCode && str "state" cd == mockState
then dat "auth.login" (object
[ "type" .= ("session" :: Text), "token" .= ("mgs_person" :: Text)
, "principal" .= ("oidc:workos/user_01M4QXZ7ABCDEF" :: Text), "name" .= ("Ada Lovelace" :: Text)
, "expires_at" .= (1791663600 :: Int)
, "memberships" .= [object ["role" .= ("admin" :: Text), "scope" .= ("*" :: Text)]] ])
else err "the code is not good for this state" []
| Just (Object s) <- KM.lookup "session" f -> dat "auth.login" (object
[ "type" .= ("session" :: Text), "token" .= ("mgs_agent_auth" :: Text)
, "principal" .= ("key:79b5" :: Text), "agent" .= ("worker-1" :: Text)
, "expires_at" .= (1791663600 :: Int)
, "ceiling" .= maybe (toJSON ([] :: [Text])) id (KM.lookup "ceiling" s) ])
_ -> err "auth.login: no flow" []
"session.login" | scSessionLogin sc -> dat "session.login" (object
[ "type" .= ("login" :: Text), "token" .= ("mgs_agent_session" :: Text)
, "principal" .= ("key:79b5" :: Text), "agent" .= ("worker-1" :: Text)
, "expires_at" .= (1791663600 :: Int)
, "scope" .= maybe (toJSON ([] :: [Text])) id (KM.lookup "scope" args) ])
other -> err ("Activation not found: " <> T.takeWhile (/= '.') other) []
-- | The browser coming back to synapse's loopback, as the provider's
-- redirect sends it.
browserReturns :: Text -> Text -> IO ()
browserReturns redirect state = do
let rest = maybe redirect id (T.stripPrefix "http://127.0.0.1:" redirect)
(portT, path) = T.breakOn "/" rest
port = read (T.unpack portT) :: Int
sock <- N.socket N.AF_INET N.Stream N.defaultProtocol
N.connect sock (N.SockAddrInet (fromIntegral port) (N.tupleToHostAddress (127, 0, 0, 1)))
-- a favicon first, as browsers do: the loopback must not take it for an answer
NB.sendAll sock "GET /favicon.ico HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n"
_ <- NB.recv sock 4096
N.close sock
sock' <- N.socket N.AF_INET N.Stream N.defaultProtocol
N.connect sock' (N.SockAddrInet (fromIntegral port) (N.tupleToHostAddress (127, 0, 0, 1)))
NB.sendAll sock' $ TE.encodeUtf8 $
"GET " <> path <> "?code=" <> mockCode <> "&state=" <> state <> " HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n"
_ <- NB.recv sock' 4096
N.close sock'
-- | RETURNS-40: a root schema with the `ir` method beside its own.
withIr :: Value -> Value
withIr (Object o) = Object (KM.insert "methods" (Array (V.snoc ms irMethod)) o)
where
ms = case KM.lookup "methods" o of { Just (Array a) -> a; _ -> mempty }
irMethod = object
[ "name" .= ("ir" :: Text), "description" .= ("The service's IR" :: Text), "hash" .= ("ir0" :: Text)
, "cardinality" .= ("one" :: Text), "returns" .= object [] ]
withIr v = v
rootSchema :: Scenario -> Value
rootSchema sc = object $
[ "namespace" .= ("menger" :: Text), "version" .= ("1.0.0" :: Text)
, "description" .= ("Central routing and introspection" :: Text)
, "hash" .= (case scView sc of { AnonymousView -> "v-anon-3c1f"; CallerView | scCatalog sc -> "rc-v9a77"; _ | scCatalog sc -> "rc0"; _ -> "r0" } :: Text)
, "children" .= children ]
where
children = case scView sc of
_ | scCatalog sc ->
[ object ["namespace" .= ("tickets" :: Text), "description" .= ("Plans as data" :: Text), "hash" .= ("t0" :: Text)]
, object ["namespace" .= ("deployments" :: Text), "description" .= ("Deployments" :: Text), "hash" .= ("d0" :: Text)] ]
AnonymousView -> [object ["namespace" .= ("auth" :: Text), "methods" .= [
object ["name" .= ("providers" :: Text), "public" .= True],
object ["name" .= ("login" :: Text), "public" .= True],
object ["name" .= ("signup" :: Text), "public" .= True]]]]
_ -> [ object ["namespace" .= ("tickets" :: Text), "description" .= ("Plans as data" :: Text), "hash" .= ("t0" :: Text)] ]
<> [ object ["namespace" .= ("auth" :: Text), "description" .= ("Sign in, who am I, what does this need" :: Text), "hash" .= ("a0" :: Text)] | scAuth sc ]
authSchema :: Value
authSchema = object
[ "namespace" .= ("auth" :: Text), "version" .= ("0.1.0" :: Text), "hash" .= ("a0" :: Text)
, "description" .= ("Sign in, who am I, what does this need" :: Text)
, "methods" .= [ m "providers" "The ways in this server offers"
, m "login" "Sign in \x2014 a person through the browser, an agent with its key"
, m "signup" "Create an account, or ask for access" ] ]
where m n d = object ["name" .= (n :: Text), "description" .= (d :: Text), "hash" .= (n <> "0"), "public" .= True, "returns" .= object []]
ticketsSchema :: Scenario -> Value
ticketsSchema sc = object $
[ "namespace" .= ("tickets" :: Text), "version" .= ("0.5.0" :: Text), "hash" .= ("t0" :: Text)
, "description" .= ("Plans as data" :: Text)
, "methods" .= map method
[ ("get", "One ticket by id", ["MENGER"], Just "id")
, ("list", "Tickets, filtered and paged", ["MENGER"], Nothing)
, ("set_status", "Change a ticket's status", ["MENGER-12"], Just "id") ] ]
<> [ "view" .= ("caller" :: Text) | caller ]
where
caller = scView sc == CallerView
method (n, d, held, target) = object $
[ "name" .= (n :: Text), "description" .= (d :: Text), "hash" .= (n <> "0"), "returns" .= object []
, "params" .= object ["type" .= ("object" :: Text), "required" .= (["id"] :: [Text]),
"properties" .= object ["id" .= object ["type" .= ("string" :: Text), "description" .= ("Ticket id, e.g. MNG-2" :: Text)]]] ]
<> (if caller then [ "grant" .= ("tickets." <> n), "held_on" .= (held :: [Text]) ] <> maybe [] (\t -> ["target" .= (t :: Text)]) target else [])
-- | SYNAPSE-28: the schema after @get@ was renamed @fetch@ (a new hash).
renameGet :: Value -> Value
renameGet (Object o) = Object (KM.insert "hash" (String "t1") (maybe o (\v -> KM.insert "methods" (ms v) o) (KM.lookup "methods" o)))
where
ms (Array a) = Array (V.map m a)
ms v = v
m (Object mo) | KM.lookup "name" mo == Just (String "get") =
Object (KM.insert "name" (String "fetch") (KM.insert "hash" (String "fetch0") mo))
m v = v
renameGet v = v
-- | The tickets schema as the caller's view shows it: without what only an
-- admin holds (tickets.delete).
catalogView :: Scenario -> Object -> Object
catalogView sc o
| scView sc /= CallerView = o
| otherwise = KM.insert "view" (String "caller") (maybe o (\ms -> KM.insert "methods" (dropDelete ms) o) (KM.lookup "methods" o))
where
dropDelete (Array ms) = Array (V.filter (\case { Object m -> KM.lookup "name" m /= Just (String "delete"); _ -> True }) ms)
dropDelete v = v
str :: Text -> Object -> Text
str k o = case KM.lookup (K.fromText k) o of
Just (String s) -> s
_ -> ""