{-# LANGUAGE PatternSynonyms #-}
-- | JSON-RPC types for communicating with Plexus RPC servers
module Plexus.Types
( -- * JSON-RPC Protocol
RpcRequest(..)
, RpcResponse(..)
, RpcError(..)
, SubscriptionNotification(..)
, SubNotifParams(..)
, RequestId(..)
, SubscriptionId(..)
-- * Plexus Stream Types
, PlexusStreamItem(.., StreamError)
, itemRefusal
, itemErrorMessage
, Refusal(..)
, Provenance(..)
, StreamMetadata(..)
, GuidanceErrorType(..)
, GuidanceSuggestion(..)
, TransportError(..)
-- * Helpers
, mkSubscribeRequest
, mkUnsubscribeRequest
, getPlexusHash
) where
import Data.Aeson
import qualified Data.Aeson.KeyMap as KM
import Data.Aeson.Types (Parser)
import qualified Data.ByteString.Lazy as LBS
import Data.Maybe (catMaybes)
import Data.Text (Text)
import qualified Data.Text as T
import qualified Data.Text.Encoding as T
import GHC.Generics (Generic)
import Plexus.Auth (Refusal(..), refusalFromObject)
-- Only the types: RemoteError's constructors share names with
-- 'GuidanceErrorType''s.
import qualified Plexus.RemoteError as RE
import Plexus.RemoteError (RemoteError)
-- | Request ID for JSON-RPC calls
newtype RequestId = RequestId { unRequestId :: Int }
deriving stock (Show, Eq, Ord)
deriving newtype (ToJSON, FromJSON)
-- | Subscription ID returned by the server (can be string or number)
newtype SubscriptionId = SubscriptionId { unSubscriptionId :: Text }
deriving stock (Show, Eq, Ord)
deriving newtype (ToJSON)
instance FromJSON SubscriptionId where
parseJSON (String s) = pure $ SubscriptionId s
parseJSON v =
-- Store raw JSON representation as the ID
pure $ SubscriptionId $ T.decodeUtf8 $ LBS.toStrict $ encode v
-- | JSON-RPC 2.0 request
data RpcRequest = RpcRequest
{ rpcReqJsonrpc :: Text -- ^ Always "2.0"
, rpcReqMethod :: Text -- ^ Method name (e.g., "bash_execute")
, rpcReqParams :: Value -- ^ Parameters (array or object)
, rpcReqId :: RequestId -- ^ Request identifier
}
deriving stock (Show, Eq, Generic)
instance ToJSON RpcRequest where
toJSON RpcRequest{..} = object
[ "jsonrpc" .= rpcReqJsonrpc
, "method" .= rpcReqMethod
, "params" .= rpcReqParams
, "id" .= rpcReqId
]
-- | JSON-RPC 2.0 response
data RpcResponse
= RpcSuccess
{ rpcRespId :: RequestId
, rpcRespResult :: Value
}
| RpcError
{ rpcRespId :: RequestId
, rpcRespError :: RpcError
}
deriving stock (Show, Eq, Generic)
instance FromJSON RpcResponse where
parseJSON = withObject "RpcResponse" $ \o -> do
rid <- o .: "id"
mResult <- o .:? "result"
mError <- o .:? "error"
case (mResult, mError) of
(Just r, Nothing) -> pure $ RpcSuccess rid r
(Nothing, Just e) -> pure $ RpcError rid e
_ -> fail "Expected either 'result' or 'error' field"
-- | JSON-RPC error object
--
-- A Plexus rejection (plexus 0.7) keeps JSON-RPC's @code@ and @message@
-- (the 'RemoteError''s one-line rendering) and carries the typed
-- 'RemoteError' in @data@:
--
-- @
-- {"code": -32601, "message": "method not found: tickets.frob",
-- "data": {"kind": "method_not_found", "activation": "tickets", "method": "frob"}}
-- @
--
-- 'errData' is 'Nothing' for a rejection that is not a Plexus call failure
-- and so carries no 'RemoteError': no @data@ (jsonrpsee's own
-- @"Method not found"@, a service's @{ns}.schema@ / @{ns}.respond@ refusals), or
-- a @data@ that is not a @kind@-tagged object (jsonrpsee's param-parse
-- rejection carries a string). A @kind@-tagged @data@ is decoded strictly: a
-- kind this library does not know is a decode error, as for an error frame.
data RpcError = RpcErrorObj
{ errCode :: Int
, errMessage :: Text
, errData :: Maybe RemoteError
}
deriving stock (Show, Eq, Generic)
instance FromJSON RpcError where
parseJSON = withObject "RpcError" $ \o -> RpcErrorObj
<$> o .: "code"
<*> o .: "message"
<*> (o .:? "data" >>= traverse parseJSON . (>>= remoteErrorShaped))
where
remoteErrorShaped :: Value -> Maybe Value
remoteErrorShaped v@(Object d) | KM.member "kind" d = Just v
remoteErrorShaped _ = Nothing
instance ToJSON RpcError where
toJSON RpcErrorObj{..} = object $
[ "code" .= errCode
, "message" .= errMessage
] <> catMaybes [("data" .=) <$> errData]
-- | Subscription notification from server
-- This is what we receive for each stream item
data SubscriptionNotification = SubscriptionNotification
{ subNotifJsonrpc :: Text
, subNotifMethod :: Text -- ^ Usually "subscription"
, subNotifParams :: SubNotifParams
}
deriving stock (Show, Eq, Generic)
data SubNotifParams = SubNotifParams
{ subParamsSubscription :: SubscriptionId
, subParamsResult :: Value
}
deriving stock (Show, Eq, Generic)
instance FromJSON SubscriptionNotification where
parseJSON = withObject "SubscriptionNotification" $ \o -> SubscriptionNotification
<$> o .: "jsonrpc"
<*> o .: "method"
<*> o .: "params"
instance FromJSON SubNotifParams where
parseJSON = withObject "SubNotifParams" $ \o -> SubNotifParams
<$> o .: "subscription"
<*> o .: "result"
-- | Provenance tracking nested calls through activations
-- Now just a list of namespace segments (no wrapper object)
newtype Provenance = Provenance { segments :: [Text] }
deriving stock (Show, Eq, Generic)
instance FromJSON Provenance where
parseJSON v = case v of
-- New format: just an array ["plexus", "echo"]
Array arr -> Provenance <$> mapM parseJSON (foldr (:) [] arr)
-- Legacy format: {"segments": [...]}
Object o -> Provenance <$> o .: "segments"
_ -> fail "Provenance: expected array or object"
instance ToJSON Provenance where
toJSON (Provenance segs) = toJSON segs
-- | Metadata wrapper for stream results
data StreamMetadata = StreamMetadata
{ metaProvenance :: Provenance
, metaPlexusHash :: Text
, metaTimestamp :: Integer
}
deriving stock (Show, Eq, Generic)
instance FromJSON StreamMetadata where
parseJSON = withObject "StreamMetadata" $ \o -> StreamMetadata
<$> o .: "provenance"
<*> o .: "plexus_hash"
<*> o .: "timestamp"
instance ToJSON StreamMetadata where
toJSON StreamMetadata{..} = object
[ "provenance" .= metaProvenance
, "plexus_hash" .= metaPlexusHash
, "timestamp" .= metaTimestamp
]
-- | Error type from guidance events
data GuidanceErrorType
= ActivationNotFound
{ errorActivation :: Text
}
| MethodNotFound
{ errorActivation :: Text
, errorMethod :: Text
}
| InvalidParams
{ errorMethod :: Text
, errorReason :: Text
}
| TransportErrorGuidance
{ errorTransport :: TransportError
}
deriving stock (Show, Eq, Generic)
instance FromJSON GuidanceErrorType where
parseJSON = withObject "GuidanceErrorType" $ \o -> do
kind <- o .: "error_kind" :: Parser Text
case kind of
"activation_not_found" -> ActivationNotFound <$> o .: "activation"
"method_not_found" -> MethodNotFound <$> o .: "activation" <*> o .: "method"
"invalid_params" -> InvalidParams <$> o .: "method" <*> o .: "reason"
"transport_error" -> TransportErrorGuidance <$> parseJSON (Object o)
_ -> fail $ "Unknown error_kind: " <> T.unpack kind
instance ToJSON GuidanceErrorType where
toJSON (ActivationNotFound activation) = object
[ "error_kind" .= ("activation_not_found" :: Text)
, "activation" .= activation
]
toJSON (MethodNotFound activation method) = object
[ "error_kind" .= ("method_not_found" :: Text)
, "activation" .= activation
, "method" .= method
]
toJSON (InvalidParams method reason) = object
[ "error_kind" .= ("invalid_params" :: Text)
, "method" .= method
, "reason" .= reason
]
toJSON (TransportErrorGuidance transportErr) =
-- Merge the transport error JSON with error_kind
case toJSON transportErr of
Object o -> Object o
_ -> object ["error_kind" .= ("transport_error" :: Text)]
-- | Suggestion for next action from guidance events
data GuidanceSuggestion
= CallPlexusSchema
| CallActivationSchema
{ suggestionNamespace :: Text
}
| TryMethod
{ suggestionMethod :: Text
, suggestionExampleParams :: Maybe Value
}
| CustomGuidance
{ suggestionMessage :: Text
}
deriving stock (Show, Eq, Generic)
instance FromJSON GuidanceSuggestion where
parseJSON = withObject "GuidanceSuggestion" $ \o -> do
action <- o .: "action" :: Parser Text
case action of
"call_plexus_schema" -> pure CallPlexusSchema -- Legacy: maps to {backend}.schema
"call_activation_schema" -> CallActivationSchema <$> o .: "namespace"
"try_method" -> TryMethod <$> o .: "method" <*> o .:? "example_params"
"custom" -> CustomGuidance <$> o .: "message"
_ -> fail $ "Unknown action: " <> T.unpack action
instance ToJSON GuidanceSuggestion where
toJSON CallPlexusSchema = object ["action" .= ("call_plexus_schema" :: Text)] -- Legacy: represents {backend}.schema
toJSON (CallActivationSchema namespace) = object
[ "action" .= ("call_activation_schema" :: Text)
, "namespace" .= namespace
]
toJSON (TryMethod method exampleParams) = object $
[ "action" .= ("try_method" :: Text)
, "method" .= method
] <> catMaybes [("example_params" .=) <$> exampleParams]
toJSON (CustomGuidance message) = object
[ "action" .= ("custom" :: Text)
, "message" .= message
]
-- ============================================================================
-- Transport Errors
-- ============================================================================
-- | Strongly-typed transport-level errors
-- These occur before reaching the backend (connection, network, protocol issues)
data TransportError
= ConnectionRefused
{ transportHost :: Text
, transportPort :: Int
}
| ConnectionTimeout
{ transportHost :: Text
, transportPort :: Int
}
| ProtocolError
{ transportMessage :: Text
}
| NetworkError
{ transportMessage :: Text
}
-- | SYNAPSE-20: an open connection was lost (the server closed it, it was
-- reset, or a keepalive ping went unanswered) while a call was on it.
-- 'transportSent' says whether the call's request had been sent: when it
-- had not, repeating the call is always safe; when it had, the server
-- may have acted on it.
| ConnectionLost
{ transportMessage :: Text
, transportSent :: Bool
}
deriving stock (Show, Eq, Generic)
instance ToJSON TransportError where
toJSON (ConnectionRefused host port) = object
[ "error_kind" .= ("connection_refused" :: Text)
, "host" .= host
, "port" .= port
]
toJSON (ConnectionTimeout host port) = object
[ "error_kind" .= ("connection_timeout" :: Text)
, "host" .= host
, "port" .= port
]
toJSON (ProtocolError msg) = object
[ "error_kind" .= ("protocol_error" :: Text)
, "message" .= msg
]
toJSON (NetworkError msg) = object
[ "error_kind" .= ("network_error" :: Text)
, "message" .= msg
]
toJSON (ConnectionLost msg sent) = object
[ "error_kind" .= ("connection_lost" :: Text)
, "message" .= msg
, "sent" .= sent
]
instance FromJSON TransportError where
parseJSON = withObject "TransportError" $ \o -> do
kind <- o .: "error_kind" :: Parser Text
case kind of
"connection_refused" -> ConnectionRefused <$> o .: "host" <*> o .: "port"
"connection_timeout" -> ConnectionTimeout <$> o .: "host" <*> o .: "port"
"protocol_error" -> ProtocolError <$> o .: "message"
"network_error" -> NetworkError <$> o .: "message"
"connection_lost" -> ConnectionLost <$> o .: "message" <*> o .: "sent"
_ -> fail $ "Unknown transport error_kind: " <> T.unpack kind
-- ============================================================================
-- Plexus Stream Items
-- ============================================================================
-- | Unified stream item from Plexus RPC
--
-- New format uses metadata wrapper:
-- @{"type":"data","metadata":{...},"content_type":"...","content":{...}}@
--
-- plexus 0.7 frames:
--
-- @
-- {"type":"progress","metadata":{...},"content_type":"ns.method","content":<P>}
-- {"type":"error","metadata":{...},"recoverable":false,"error":<RemoteError>}
-- @
--
-- A progress frame carries the method's declared progress type @P@ (the
-- schema's 'Plexus.Schema.Recursive.methodProgress'; for the default @Note@,
-- @{"message": …, "percentage": …}@). An error frame means the call itself
-- failed; an activation saying no is a 'StreamData' item whose content is
-- @{"err": …}@.
data PlexusStreamItem
= StreamProgress
{ itemPlexusHash :: Text
, itemProvenance :: Provenance
, itemContentType :: Text -- ^ The call's method, as on its data items (e.g. @"tickets.import"@)
, itemContent :: Value -- ^ The method's progress value @P@
}
| StreamData
{ itemPlexusHash :: Text
, itemProvenance :: Provenance
, itemContentType :: Text
, itemData :: Value
}
| StreamGuidance
{ itemPlexusHash :: Text
, itemProvenance :: Provenance
, itemErrorType :: GuidanceErrorType
, itemSuggestion :: GuidanceSuggestion
, itemAvailableMethods :: Maybe [Text]
, itemMethodSchema :: Maybe Value
}
| StreamErrorWith
{ itemPlexusHash :: Text
, itemProvenance :: Provenance
, itemError :: RemoteError -- ^ Why the call failed, typed
, itemRecoverable :: Bool -- ^ Whether the operation can be retried
, itemErrorRefusal :: Maybe Refusal
-- ^ AUTH-2: the refusal's own fields (@error_kind: "forbidden"@,
-- @grant@, @scope@, @roles@, @held_on@) when the service sends them
-- beside the typed error; Nothing otherwise. 'itemRefusal' also
-- reads a refusal out of a typed @Forbidden (MissingGrant …)@.
, itemErrorBeside :: KM.KeyMap Value
-- ^ RETURNS-36: every other key the frame carried beside the typed
-- error (a merged service's @message@, @code@, @error_kind@ and
-- the refusal's fields), kept as sent and written back so; empty
-- for a frame built here.
}
| StreamDone
{ itemPlexusHash :: Text
, itemProvenance :: Provenance
}
deriving stock (Show, Eq, Generic)
-- | An error item: hash, provenance, the typed 'RemoteError', recoverable.
-- Matches every error (a refusal too); builds one with no refusal fields.
-- 'StreamErrorWith' carries AUTH-2's refusal fields when a service sends them.
pattern StreamError :: Text -> Provenance -> RemoteError -> Bool -> PlexusStreamItem
pattern StreamError hash prov err rec <- StreamErrorWith hash prov err rec _ _
where StreamError hash prov err rec = StreamErrorWith hash prov err rec Nothing KM.empty
{-# COMPLETE StreamProgress, StreamData, StreamGuidance, StreamError, StreamDone #-}
-- | AUTH-2: the refusal an item carries — an error item with
-- @error_kind: "forbidden"@ (or whose typed error is a
-- @Forbidden (MissingGrant …)@, plexus 0.8), or a data item whose content is
-- one (an in-band error event naming the same fields).
itemRefusal :: PlexusStreamItem -> Maybe Refusal
itemRefusal = \case
StreamErrorWith _ _ _ _ (Just r) _ -> Just r
StreamErrorWith _ _ err _ Nothing _ -> refusalOfRemote err
StreamData _ _ _ (Object o) -> refusalFromObject o
_ -> Nothing
instance FromJSON PlexusStreamItem where
parseJSON = withObject "PlexusStreamItem" $ \o -> do
typ <- o .: "type" :: Parser Text
-- New format: metadata is a nested object
mMeta <- o .:? "metadata" :: Parser (Maybe StreamMetadata)
case mMeta of
Just meta -> parseWithMetadata typ meta o
Nothing -> parseLegacy typ o
where
-- New format: metadata wrapper
parseWithMetadata :: Text -> StreamMetadata -> Object -> Parser PlexusStreamItem
parseWithMetadata typ meta o = do
let hash = metaPlexusHash meta
prov = metaProvenance meta
case typ of
"progress" -> StreamProgress hash prov
<$> o .:? "content_type" .!= ""
<*> progressContent o
"data" -> StreamData hash prov
<$> o .: "content_type"
<*> o .: "content"
"guidance" -> StreamGuidance hash prov
<$> o .: "error_type"
<*> o .: "suggestion"
<*> o .:? "available_methods"
<*> o .:? "method_schema"
"error" -> StreamErrorWith hash prov
<$> typedError o
<*> o .: "recoverable"
<*> pure (refusalFromObject o)
<*> pure (besideError o)
"done" -> pure $ StreamDone hash prov
_ -> fail $ "Unknown event type: " <> T.unpack typ
-- Legacy format: flat structure
parseLegacy :: Text -> Object -> Parser PlexusStreamItem
parseLegacy typ o = do
-- an error item may come bare (AUTH-2's refusal is written so):
-- no hash or provenance is no reason to lose it
hash <- if typ == "error" then o .:? "plexus_hash" .!= "" else o .: "plexus_hash"
case typ of
"progress" -> StreamProgress hash
<$> o .: "provenance"
<*> o .: "content_type"
<*> o .: "content"
"data" -> StreamData hash
<$> o .: "provenance"
<*> o .: "content_type"
<*> o .: "data"
"guidance" -> StreamGuidance hash
<$> o .: "provenance"
<*> o .: "error_type"
<*> o .: "suggestion"
<*> o .:? "available_methods"
<*> o .:? "method_schema"
"error" -> StreamErrorWith hash
<$> o .:? "provenance" .!= Provenance []
<*> typedError o
<*> o .:? "recoverable" .!= False
<*> pure (refusalFromObject o)
<*> pure (besideError o)
"done" -> StreamDone hash
<$> o .: "provenance"
_ -> fail $ "Unknown event type: " <> T.unpack typ
instance ToJSON PlexusStreamItem where
toJSON (StreamProgress hash prov ct content) = object
[ "type" .= ("progress" :: Text)
, "metadata" .= StreamMetadata prov hash 0
, "content_type" .= ct
, "content" .= content
]
toJSON (StreamData hash prov ct dat) = object
[ "type" .= ("data" :: Text)
, "metadata" .= StreamMetadata prov hash 0
, "content_type" .= ct
, "content" .= dat
]
toJSON (StreamGuidance hash prov errorType suggestion availMethods methodSchema) = object $
[ "type" .= ("guidance" :: Text)
, "metadata" .= StreamMetadata prov hash 0
, "error_type" .= errorType
, "suggestion" .= suggestion
] <> catMaybes
[ ("available_methods" .=) <$> availMethods
, ("method_schema" .=) <$> methodSchema
]
toJSON (StreamErrorWith hash prov err rec refusal beside) = object $
[ "type" .= ("error" :: Text)
, "metadata" .= StreamMetadata prov hash 0
, "recoverable" .= rec
, "error" .= err
] <> (if KM.member "message" beside then [] else ["message" .= RE.renderRemoteError err])
<> if KM.null beside then refusalFields else KM.toList beside
where
-- a frame built here with a refusal: its own fields, beside the typed error
refusalFields = case toJSON <$> refusal of
Just (Object ro) -> [ (k, v) | (k, v) <- KM.toList ro, k /= "message" ]
_ -> []
toJSON (StreamDone hash prov) = object
[ "type" .= ("done" :: Text)
, "metadata" .= StreamMetadata prov hash 0
]
-- | RETURNS-36 review: the server's own line for an error frame — its
-- @message@, as plexus-core writes it on every error frame (and the 0.7.0
-- line did) — when the frame came with one; 'Nothing' for a frame built
-- here or sent without it. A caller shows it, and reads it, as main did.
itemErrorMessage :: PlexusStreamItem -> Maybe Text
itemErrorMessage = \case
StreamErrorWith _ _ _ _ _ beside | Just (String m) <- KM.lookup "message" beside -> Just m
_ -> Nothing
-- | An error frame's typed error: its @error@ (plexus 0.8), else — a frame
-- of the published 0.7.0 line — what its spec/05 fields say
-- ('RE.remoteErrorOfFields'; RETURNS-36).
typedError :: Object -> Parser RemoteError
typedError o = case KM.lookup "error" o of
Just v -> parseJSON v
Nothing -> pure (RE.remoteErrorOfFields o)
-- | A progress frame's content: its @content@ (plexus 0.8), else — a 0.7.0
-- frame — its @message@ and @percentage@ as the default progress type,
-- @Note@ (RETURNS-36, as plexus-core's child seam reads it).
progressContent :: Object -> Parser Value
progressContent o = case KM.lookup "content" o of
Just v -> pure v
Nothing -> pure $ Object $ KM.fromList $
("message", maybe (String "") id (KM.lookup "message" o))
: [ ("percentage", p) | Just p <- [KM.lookup "percentage" o], p /= Null ] -- a null one left out, as spec05.rs
-- | An error frame's keys beside the envelope and the typed error.
besideError :: Object -> KM.KeyMap Value
besideError = KM.filterWithKey (\k _ -> k `notElem` ["type", "metadata", "provenance", "plexus_hash", "recoverable", "error"])
-- | The AUTH-2 refusal a typed @Forbidden (MissingGrant …)@ names.
refusalOfRemote :: RemoteError -> Maybe Refusal
refusalOfRemote = \case
RE.Forbidden (RE.MissingGrant g sc roles held) -> Just Refusal
{ refusalGrant = Just g
, refusalScope = Just sc
, refusalRoles = roles
, refusalHeldOn = held
, refusalCeiling = []
, refusalMessage = if sc == "*" then "needs " <> g else "needs " <> g <> " on " <> sc
}
_ -> Nothing
-- | Create a subscription request
mkSubscribeRequest :: RequestId -> Text -> Value -> RpcRequest
mkSubscribeRequest rid method params = RpcRequest
{ rpcReqJsonrpc = "2.0"
, rpcReqMethod = method
, rpcReqParams = params
, rpcReqId = rid
}
-- | Create an unsubscribe request
mkUnsubscribeRequest :: RequestId -> Text -> SubscriptionId -> RpcRequest
mkUnsubscribeRequest rid unsubMethod subId = RpcRequest
{ rpcReqJsonrpc = "2.0"
, rpcReqMethod = unsubMethod
, rpcReqParams = toJSON [unSubscriptionId subId]
, rpcReqId = rid
}
-- | Extract the Plexus RPC hash from a stream item
getPlexusHash :: PlexusStreamItem -> Text
getPlexusHash = itemPlexusHash