path: root/Presence/XMPP.hs

{-# LANGUAGE OverloadedStrings   #-}
{-# LANGUAGE FlexibleContexts    #-}
{-# LANGUAGE CPP                 #-}
{-# LANGUAGE ViewPatterns        #-}
module XMPP 
    ( module XMPPTypes
    , module SocketLike
    , listenForXmppClients
    , listenForRemotePeers
    , newServerConnections
    , seekRemotePeers
    , quitListening
    , OutBoundMessage(..)
    , sendMessage
    )  where

import ServerC
import XMPPTypes
import SocketLike
import ByteStringOperators
import ControlMaybe

import Data.Maybe (catMaybes)
import Data.HList
import Network.Socket 
    ( Family
    , connect
    , socketToHandle
    , sClose
    , Socket(..)
    , socket
    , SocketType(..)
    )
import Network.BSD 
    ( PortNumber
    , getHostName
    , hostName
    , hostAliases
    , getProtocolNumber
    )
import System.IO
    ( BufferMode(..)
    , IOMode(..)
    , hSetBuffering
    )
import Control.Concurrent.STM
import Data.Conduit
import qualified Data.Conduit.List as CL
import qualified Data.Conduit.Binary as CB
import Data.ByteString (ByteString)
import qualified Data.ByteString.Char8 as S (pack,putStr,putStrLn,append)
import qualified Data.ByteString.Lazy.Char8 as L
    ( putStrLn
    , fromChunks
    , unlines
    , hPutStrLn
    )
import Control.Concurrent (forkIO,killThread)
import Control.Concurrent.Async
import Control.Exception 
    ( handle
    -- , SomeException(..)
    , finally
    , bracketOnError )
import GHC.IO.Exception (IOException(..))
import System.IO.Error (isDoesNotExistError)
import Control.Monad.IO.Class
import Control.Monad.Trans.Class
import Control.Monad.Trans.Maybe
import Todo
import Control.Monad as Monad
import Text.XML.Stream.Parse (parseBytes,content)
import Text.XML.Stream.Render
import Data.XML.Types as XML
import qualified Data.Text as S (takeWhile)
import Data.Text.Encoding      as S (decodeUtf8,encodeUtf8)
import Data.Text.Lazy.Encoding as L (decodeUtf8,encodeUtf8)
import Data.Text.Lazy (toStrict)
import Data.Monoid
import qualified Data.Sequence as Seq
import Data.Foldable (toList)
#ifdef RENDERFLUSH
import Data.Conduit.Blaze
#endif
import Data.List (find)
import qualified Text.Show.ByteString as L
import NestingXML
import           Data.Set as Set (Set,(\\))
import qualified Data.Set as Set
import qualified Data.Map as Map
import           Data.Map as Map (Map)
import GHC.Conc 
    ( threadStatus
    , ThreadStatus(..)
    , ThreadId
    )

textToByteString x = L.fromChunks [S.encodeUtf8 x]


data Commands = 
    Send [XML.Event] 
    | BoundToResource
    | InterestedInRoster
    | QuitThread 
 deriving Prelude.Show


xmlifyPresenceForClient :: Presence -> IO [XML.Event]
xmlifyPresenceForClient (Presence jid stat) = do
    let n   = name jid
        rsc = resource jid
    names <- getNamesForPeer (peer jid)
    let tostr p = L.decodeUtf8 $ n <$++> "@" <?++> L.fromChunks [p] <++?>  "/" <++$> rsc
        jidstrs = fmap (toStrict . tostr) names
    return (concatMap presenceEvents jidstrs)
 where
    presenceEvents jidstr =
      [ EventBeginElement "{jabber:client}presence" (("from",[ContentText jidstr]):typ stat)
      , EventBeginElement "{jabber:client}show" []
      , EventContent (ContentText . shw $ stat)
      , EventEndElement "{jabber:client}show"
      , EventEndElement "{jabber:client}presence"
      ]
    typ Offline = [("type",[ContentText "unavailable"])]
    typ _       = []
    shw Available = "chat"
    shw Away      = "away"
    shw Offline   = "away" -- Is this right?

prefix ## name = Name name Nothing (Just prefix)

streamP name = Name name (Just "http://etherx.jabber.org/streams") (Just "stream")
    
greet host =
    [ EventBeginDocument
    , EventBeginElement (streamP "stream")
        [("from",[ContentText host])
        ,("id",[ContentText "someid"])
        ,("xmlns",[ContentText "jabber:client"])
        ,("xmlns:stream",[ContentText "http://etherx.jabber.org/streams"])
        ,("version",[ContentText "1.0"])
        ]
    , EventBeginElement (streamP "features") []
    , EventBeginElement "{urn:ietf:params:xml:ns:xmpp-bind}bind" []
    , EventEndElement "{urn:ietf:params:xml:ns:xmpp-bind}bind"

    {-
    -- , "  <session xmlns='urn:ietf:params:xml:ns:xmpp-session'/>"
    , " <mechanisms xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>"
    -- , " <mechanism>DIGEST-MD5</mechanism>"
    , " <mechanism>PLAIN</mechanism>"
    , " </mechanisms> "
    -}

    , EventEndElement (streamP "features")
    ]


-- type Consumer i m r = forall o. ConduitM i o m r
mawait :: Monad m => MaybeT (ConduitM i o m) i
mawait = MaybeT await

-- Note: This function ignores name space qualification
elementAttrs expected (EventBeginElement name attrs)
    | nameLocalName name==expected 
    = return attrs
elementAttrs _ _ = mzero
     
eventIsBeginElement (EventBeginElement _ _) = True
eventIsBeginElement _                       = False

eventIsEndElement (EventEndElement _) = True
eventIsEndElement _                   = False

filterMapElement::
  (Monad m, MonadPlus mp) =>
  (Event -> mp a) -> Event -> mp a -> MaybeT (ConduitM Event o m) (mp a)
filterMapElement ret opentag empty = loop (empty `mplus` ret opentag) 1 
 where
    loop ts 0 = return ts
    loop ts cnt = do
        tag <- mawait
        let ts' = mplus ts (ret tag)
        case () of
         _ | eventIsEndElement tag   -> loop ts' (cnt-1)
         _ | eventIsBeginElement tag -> loop ts' (cnt+1)
         _                           -> loop ts' cnt

gatherElement ::
  (Monad m, MonadPlus mp) =>
  Event -> mp Event -> NestingXML o m (mp Event)
gatherElement opentag empty = loop (empty `mplus` return opentag) 1 
 where
    loop ts 0 = return ts
    loop ts cnt = do
        maybeXML (return ts) $ \tag -> do
        let ts' = mplus ts (return tag)
        case () of
         _ | eventIsEndElement tag   -> loop ts' (cnt-1)
         _ | eventIsBeginElement tag -> loop ts' (cnt+1)
         _                           -> loop ts' cnt

{-
sourceStanza :: Monad m => Event -> ConduitM Event Event m ()
sourceStanza opentag = yield opentag >> loop 1
 where
    loop 0 = return ()
    loop cnt = do
        e <- await
        let go tag cnt = yield tag >> loop cnt
        case e of
            Just tag | eventIsEndElement tag   -> go tag (cnt-1)
            Just tag | eventIsBeginElement tag -> go tag (cnt+1)
            Just tag                           -> go tag cnt
            Nothing                            -> return ()
-}

voidMaybeT body = (>> return ()) . runMaybeT $ body
fixMaybeT f = (>> return ()) . runMaybeT . fix $ f

iq_bind_reply id jid =
    [ EventBeginElement "{jabber:client}iq" [("type",[ContentText "result"]),("id",[ContentText id])]

    , EventBeginElement "{urn:ietf:params:xml:ns:xmpp-bind}bind"
        [("xmlns",[ContentText "urn:ietf:params:xml:ns:xmpp-bind"])]
    , EventBeginElement "{urn:ietf:params:xml:ns:xmpp-bind}jid" []
    , EventContent (ContentText jid)
    , EventEndElement "{urn:ietf:params:xml:ns:xmpp-bind}jid"
    , EventEndElement  "{urn:ietf:params:xml:ns:xmpp-bind}bind"
    , EventEndElement "{jabber:client}iq"
    ]

uncontent cs = head $ map getText cs
 where
    getText (ContentText x) = x
    getText (ContentEntity x ) = x

tagAttrs (EventBeginElement _ xs) = xs
tagAttrs _                        = []

tagName (EventBeginElement n _) = n
tagName _                       = ""

handleIQSetBind session cmdChan stanza_id = do
    mchild <- nextElement
    rsc <- case mchild of
             Just child -> do
                let unhandledBind = do
                        liftIO $ putStrLn $ "unhandled-bind: "++show child
                        return ""
                case tagName child of
                    "{urn:ietf:params:xml:ns:xmpp-bind}resource"
                      -> do
                        rsc <- lift content
                        return . textToByteString $ rsc
                    _ -> unhandledBind
             Nothing -> do
                liftIO $ putStrLn $ "empty bind request!"
                return ""
    liftIO $ do
        L.putStrLn $ "iq-set-bind-resource " <++> rsc
        setResource session rsc
        jid <- getJID session
        atomically $ do
            writeTChan cmdChan (Send $ iq_bind_reply stanza_id (toStrict $ L.decodeUtf8 $ L.show jid) )
            writeTChan cmdChan BoundToResource
        forCachedPresence session $ \presence -> do
            xs <- xmlifyPresenceForClient presence
            atomically . writeTChan cmdChan . Send $ xs


iq_session_reply host stanza_id =
    [ EventBeginElement "{jabber:client}iq" 
        [("id",[ContentText stanza_id])
        ,("from",[ContentText host])
        ,("type",[ContentText "result"])
        ]
    , EventEndElement "{jabber:client}iq"
    ]

handleIQSetSession session cmdChan stanza_id = do
    host <- liftIO $ do
        jid <- getJID session
        names <- getNamesForPeer (peer jid)
        return (S.decodeUtf8 . head $ names)
    liftIO . atomically . writeTChan cmdChan . Send $ iq_session_reply host stanza_id

handleIQSet session cmdChan tag = do
    withJust (lookupAttrib "id" (tagAttrs tag)) $ \stanza_id -> do
    whenJust nextElement $ \child -> do
    let unhandledSet = liftIO $ putStrLn ("iq-set: "++show (stanza_id,child))
    case tagName child of
      "{urn:ietf:params:xml:ns:xmpp-bind}bind" 
        -> handleIQSetBind session cmdChan stanza_id
      "{urn:ietf:params:xml:ns:xmpp-session}session"
        -> handleIQSetSession session cmdChan stanza_id
      _ -> unhandledSet

matchAttrib name value attrs = 
    case find ( (==name) . fst) attrs of
        Just (_,[ContentText x])   | x==value -> True
        Just (_,[ContentEntity x]) | x==value -> True
        _                                     -> False

lookupAttrib name attrs =
    case find ( (==name) . fst) attrs of
        Just (_,[ContentText x])   -> Just x
        Just (_,[ContentEntity x]) -> Just x
        _                          -> Nothing
    
iqTypeSet    = "set"
iqTypeGet    = "get"
iqTypeResult = "result"
iqTypeError  = "error"

isIQOf (EventBeginElement name attrs) testType 
    | name=="{jabber:client}iq" 
    && matchAttrib "type" testType attrs
    = True
isIQOf _ _ = False

iq_service_unavailable host iq_id mjid req =
    [ EventBeginElement "{jabber:client}iq" 
        [("type",[ContentText "error"])
        ,("id",[ContentText iq_id])
        -- , TODO: set "from" if isJust mjid
        ]
    , EventBeginElement req []
    , EventEndElement req
    , EventBeginElement "{jabber:client}error" [("type",[ContentText "cancel"])]
    , EventBeginElement "{urn:ietf:params:xml:ns:xmpp-stanzas}service-unavailable" []
    , EventEndElement "{urn:ietf:params:xml:ns:xmpp-stanzas}service-unavailable"
    , EventEndElement "{jabber:client}error"
    , EventEndElement "{jabber:client}iq"
    ]

attr   name value = (name,[ContentText value])
attrbs name value = (name,[ContentText (toStrict . L.decodeUtf8 $ value)])


getRoster session iqid = do
    let getlist f = do
            bs <- f session
            -- js <- mapM parseHostNameJID bs
            return (Set.fromList bs) -- js)
    buddies     <- getlist getMyBuddies
    subscribers <- getlist getMySubscribers
    subnone     <- getlist getMyOthers
    solicited   <- getlist getMySolicited
    let subto = buddies \\ subscribers
    let subfrom = subscribers \\ buddies
    let subboth = Set.intersection buddies subscribers
	-- solicited -> ask='subscribe'
    jid <- getJID session
    let dest = toStrict . L.decodeUtf8 . bshow $ jid
    let items= (xmlify solicited "to" subto) 
               ++(xmlify solicited "from" subfrom)
               ++(xmlify solicited "both" subboth)
               ++(xmlify solicited "none" subnone)
        openiq =  [EventBeginElement "{jabber:client}iq"
                    [ attr "id" iqid
                    , attr "to" dest
                    , attr "type" "result" ]
                  ,EventBeginElement "{jabber:iq:roster}query"
                    [] -- todo: ver? 
                  ]
        closeiq = [EventEndElement "{jabber:iq:roster}query"
                  ,EventEndElement "{jabber:client}iq"]
    return $ openiq ++ items ++ closeiq
 where
    xmlify solicited stype set = flip concatMap (Set.toList set) 
      $ \jid -> 
        [ EventBeginElement "item" 
            ([ attr "jid" (toStrict . L.decodeUtf8 $ jid)
             , attr "subscription" stype
             ]++if Set.member jid solicited 
                 then [attr "ask" "subscribe"]
                 else [] )
        , EventEndElement "item"
        ] 

handleIQGet session cmdChan tag = do
    withJust (lookupAttrib "id" (tagAttrs tag)) $ \stanza_id -> do
    whenJust nextElement $ \child -> do
    host <- liftIO $ do
        jid <- getJID session
        names <- getNamesForPeer (peer jid)
        return (S.decodeUtf8 . head $ names)
    let unhandledGet req = do
            liftIO $ putStrLn ("iq-get: "++show (stanza_id,child))
            liftIO . atomically . writeTChan cmdChan . Send $ iq_service_unavailable host stanza_id Nothing req
    case tagName child of
        -- "{http://jabber.org/protocol/disco#items}query" -> liftIO $ putStrLn "iq-get-query-items"
        "{urn:xmpp:ping}ping" -> liftIO $ do
            let mjid = lookupAttrib "from" (tagAttrs tag)
            let pong = [ EventBeginElement "{jabber:client}iq" 
                         $ (case mjid of
                             Just jid -> (attr "to" jid :)
                             _        -> id               )
                             [ attr "type" "result"
                             , attr "id"   stanza_id
                             , attr "from" host
                             ]
                      , EventEndElement "{jabber:client}iq"
                      ]
            atomically . writeTChan cmdChan . Send $ pong
        "{jabber:iq:roster}query" -> liftIO $ do
            putStrLn $ "REQUESTED ROSTER " ++ show tag
            -- REQUESTED ROSTER EventBeginElement 
            -- (Name {nameLocalName = "iq", nameNamespace = Just "jabber:client", namePrefix = Nothing}) 
            -- [(Name { nameLocalName = "id"
            --        , nameNamespace = Nothing
            --        , namePrefix = Nothing},
            --  [ContentText "32a337c2-7b22-45b6-9d21-15ded0d079ec"])
            -- ,(Name {nameLocalName = "type", nameNamespace = Nothing, namePrefix = Nothing},
            --  [ContentText "get"])]
            roster <- getRoster session stanza_id
            atomically $ do
                writeTChan cmdChan InterestedInRoster
                writeTChan cmdChan . Send $ roster
            sendPending session
        req -> unhandledGet req
    

fromClient :: (MonadThrow m,MonadIO m, JabberClientSession session) => 
    session -> TChan Commands -> Sink XML.Event m ()
fromClient session cmdChan = doNestingXML $ do
    let log  = liftIO . L.putStrLn . ("(C) " <++>)
        send = liftIO . atomically . writeTChan cmdChan . Send
    withXML $ \begindoc -> do
    when (begindoc==EventBeginDocument) $ do
    log "begin-doc"
    withXML $ \xml -> do
    withJust (elementAttrs "stream" xml) $ \stream_attrs -> do
    log $ "stream atributes: " <++> bshow stream_attrs
    host <- liftIO $ do
        jid <- getJID session
        names <- getNamesForPeer (peer jid)
        return (S.decodeUtf8 . head $ names)
    send $ greet host
    
    fix $ \loop -> do
        log "waiting for stanza."
        whenJust nextElement $ \stanza -> do
            stanza_lvl <- nesting

            liftIO . putStrLn $ "stanza: "++show stanza

            let unhandledStanza = do
                    xs <- gatherElement stanza Seq.empty
                    prettyPrint "unhandled-C: " (toList xs)
            case () of
             _ | stanza `isIQOf` iqTypeSet -> handleIQSet session cmdChan stanza
             _ | stanza `isIQOf` iqTypeGet -> handleIQGet session cmdChan stanza
             _ | stanza `isClientPresenceOf` presenceTypeSubscribe 
                                           -> clientRequestsSubscription session cmdChan stanza
             _ | stanza `isClientPresenceOf` presenceTypeSubscribed
                                           -> clientApprovesSubscription session stanza
             _ | otherwise                 -> unhandledStanza

            awaitCloser stanza_lvl
            loop

    log $ "end of stream"
    withXML $ \xml -> do
    log $ "end-of-document: " <++> bshow xml

prettyPrint prefix xs = 
    liftIO $ do
        CL.sourceList xs $= renderBytes (def { rsPretty=True }) =$= CB.lines $$ CL.mapM_ (S.putStrLn . (prefix `S.append`))


rosterPush to contact attrs = do
    let n   = name to
        rsc = resource to
    names <- getNamesForPeer (peer to)
    let tostr p = L.decodeUtf8 $ n <$++> "@" <?++> L.fromChunks [p] <++?>  "/" <++$> rsc
        jidstrs = fmap (toStrict . tostr) names
        tojid = head jidstrs
    return
      [ EventBeginElement "{jabber:client}iq" 
            [ attr "to"   tojid
            , attr "id"   "someid"
            , attr "type" "set"
            ]
      , EventBeginElement "{jabber:iq:roster}query" []
      , EventBeginElement "{jabber:iq:roster}item" (attr "jid" contact : attrs)
      , EventEndElement   "{jabber:iq:roster}item"
      , EventEndElement   "{jabber:iq:roster}query"
      , EventEndElement "{jabber:client}iq"
      ]

data EventsForClient = CmdChan Commands 
                     | PChan   Presence
                     | RChan   RosterEvent

toClient :: (MonadIO m, JabberClientSession session ) => 
    session -> TChan Presence -> TChan Commands -> TChan RosterEvent -> Source m [XML.Event]
toClient session pchan cmdChan rchan = toClient' False False 
 where
  toClient' isBound isInterested = do
    let loop = toClient' isBound isInterested
        send xs = yield xs >> prettyPrint ">C: " xs
    event <- liftIO . atomically $ 
              foldr1 orElse [fmap CmdChan $ readTChan cmdChan
                            ,fmap RChan   $ readTChan rchan
                            ,fmap PChan   $ readTChan pchan
                            ]
    case event of
      CmdChan QuitThread -> return ()
      CmdChan (Send xs)  -> send xs >> loop
      CmdChan BoundToResource    -> toClient' True    isInterested
      CmdChan InterestedInRoster -> do
        liftIO . putStrLn $ "Roster: interested"
        toClient' isBound True
      -- CmdChan cmd        -> liftIO (putStrLn $ "unhandled event: "++show cmd) >> loop
      RChan   (RequestedSubscription who contact) ->  do
        jid <- liftIO $ getJID session
        when (isInterested && Just who==name jid) $ do
            r <- liftIO $ rosterPush jid (toStrict . L.decodeUtf8 $ contact) [attr "ask" "subscribe"]
            send r
        loop
      RChan   (NewBuddy who contact) ->  do
        liftIO . putStrLn $ "Roster push: NewBuddy "++show (isInterested,who,contact)
        (jid,me) <- liftIO $ do
                        jid <- getJID session
                        me <- asHostNameJID jid
                        return (jid,me)
        withJust me $ \me -> do
        when (isInterested && Just who==name jid) $ do
            send [ EventBeginElement "{jabber:client}presence" 
                    [ attrbs "from" contact
                    , attrbs "to"   me
                    , attr   "type" "subscribed"
                    ]
                 , EventEndElement "{jabber:client}presence" ]
            let f True  = "from"
                f False = "to"
            subscription <- fmap f (liftIO $ isSubscribed session contact)
            r <- liftIO . handleIO (\e -> putStrLn ("Roster NewBuddy error: "++show e) >> return []) $ do
                            rosterPush jid 
                                     (toStrict . L.decodeUtf8 $ contact) 
                                     [attr "subscription" subscription]
            send r
        loop
      RChan   (PendingBuddy who contact) ->  do
        liftIO . putStrLn $ "Roster: Pending buddy "++show (isInterested,who,contact)
        (jid,me) <- liftIO $ do
                        jid <- getJID session
                        me <- asHostNameJID jid
                        return (jid,me)
        withJust me $ \me -> do
        when (isInterested && Just who==name jid) $ do
            send [ EventBeginElement "{jabber:client}presence" 
                    [ attrbs "from" contact
                    , attrbs "to"   me
                    , attr   "type" "subscribe"
                    ]
                 , EventEndElement "{jabber:client}presence" ]
        loop
      PChan presence   -> do 
        when isBound $ do
            xs <- liftIO $ xmlifyPresenceForClient presence
            send xs
        loop


handleClient
  :: (SocketLike sock, HHead l (XMPPClass session),
      JabberClientSession session) =>
     HCons sock (HCons t l) -> Source IO ByteString -> Sink ByteString IO () -> IO ()
handleClient st src snk = do
    let HCons sock (HCons _ st') = st
        session_factory = hHead st'
    pname <- getPeerName sock
    session <- newSession session_factory sock
    Prelude.putStrLn $ "PEER NAME: "++Prelude.show pname
    pchan <- subscribe session Nothing
    rchan <- subscribeToRoster session
    cmdChan <- atomically newTChan

#ifdef RENDERFLUSH
    writer <- async (   toClient session pchan cmdChan rchan
                    $$  flushList
                    =$= renderBuilderFlush def 
                    =$= builderToByteStringFlush 
                    =$= discardFlush 
                     =$ snk )
#else
    writer <- async ( toClient session pchan cmdChan rchan $$ renderChunks =$ snk )
#endif
    finally         ( src $= parseBytes def $$ fromClient session cmdChan )
        $ do
            atomically $ writeTChan cmdChan QuitThread
            wait writer
            closeSession session

listenForXmppClients ::
  (HList l, HHead l (XMPPClass session), HExtend e1 l2 l1,
   HExtend e l1 (HCons PortNumber l), JabberClientSession session) =>
  Family -> e1 -> e -> l2 -> IO ServerHandle
listenForXmppClients addr_family session_factory port st = do
    doServer (addr_family .*. port .*. session_factory .*. st) handleClient

#ifdef RENDERFLUSH
flushList :: Monad m => ConduitM [a] (Flush a) m ()
flushList = fixMaybeT $ \loop -> do
    xs <- mawait
    lift ( CL.sourceList xs $$ CL.mapM_ (yield . Chunk) )
    lift ( yield Flush )
    loop

discardFlush :: Monad m => ConduitM (Flush a) a m ()
discardFlush = fixMaybeT $ \loop -> do
    x <- mawait
    let unchunk (Chunk a) = a
        ischunk (Chunk _) = True
        ischunk _         = False
    lift . when (ischunk x) $ yield (unchunk x)
    loop
#else
renderChunks :: (MonadUnsafeIO m, MonadIO m) => ConduitM [Event] ByteString m ()
renderChunks = fixMaybeT $  \loop -> do
    xs <- mawait
    lift . when (not . null $ xs) $ ( CL.sourceList xs $= renderBytes def $$ CL.mapM_ yield )
    loop
#endif


listenForRemotePeers
  :: (HList l, HHead l (XMPPPeerClass session),
      HExtend e l1 (HCons PortNumber l), HExtend e1 l2 l1,
      JabberPeerSession session) =>
     Family -> e1 -> e -> l2 -> IO ServerHandle
listenForRemotePeers addrfamily session_factory port st = do
    doServer (addrfamily .*. port .*. session_factory .*. st) handlePeer

handlePeer
  :: (SocketLike sock, HHead l (XMPPPeerClass session),
      JabberPeerSession session) =>
     HCons sock (HCons t1 l) -> Source IO ByteString -> t -> IO ()
handlePeer st src snk = do
    let HCons sock (HCons _ st') = st
        session_factory = hHead st'
    name <- fmap bshow $ getPeerName sock
    L.putStrLn $ "(P) connected " <++> name
    session <- newPeerSession session_factory sock

    finally ( src $= parseBytes def $$ fromPeer session )
        $ do 
            L.putStrLn $ "(P) disconnected " <++> name
            closePeerSession session


handlePeerPresence session stanza False = do
    -- Offline
    withJust (lookupAttrib "from" (tagAttrs stanza)) $ \jid -> do
    peer_jid <- liftIO $ parseAddressJID (textToByteString jid)
    liftIO $ announcePresence session (Presence peer_jid Offline)
handlePeerPresence session stanza True = do
    -- online (Available or Away)
    let log  = liftIO . L.putStrLn . ("(P) " <++>)
    withJust (lookupAttrib "from" (tagAttrs stanza)) $ \jid -> do
    pjid <- liftIO $ parseAddressJID (textToByteString jid)
    -- stat <- show element content
    let parseChildren stat = do
            child <- nextElement
            case child of
                Just tag | tagName tag=="{jabber:server}show"
                                      -> fmap toStat (lift content)
                Just tag | otherwise  -> parseChildren stat
                Nothing -> return stat
        toStat "away" = Away
        toStat "xa"   = Away -- TODO: xa
        toStat "dnd"  = Away -- TODO: dnd
        toStat "chat" = Available

    stat' <- parseChildren Available
    liftIO $ announcePresence session (Presence pjid stat')
    log $ bshow (Presence pjid stat')

matchAttribMaybe name (Just value) attrs = 
    case find ( (==name) . fst) attrs of
        Just (_,[ContentText x])   | x==value -> True
        Just (_,[ContentEntity x]) | x==value -> True
        _                                     -> False
matchAttribMaybe name Nothing attrs  
    | find ( (==name) . fst) attrs==Nothing
    = True
matchAttribMaybe name Nothing attrs  
    | otherwise
    = False

presenceTypeOffline = Just "unavailable"
presenceTypeOnline  = Nothing
presenceTypeProbe   = Just "probe"
presenceTypeSubscribe = Just "subscribe"
presenceTypeSubscribed = Just "subscribed"

isPresenceOf (EventBeginElement name attrs) testType
    | name=="{jabber:server}presence"
    && matchAttribMaybe "type" testType attrs
    = True
isPresenceOf _ _ = False

isClientPresenceOf (EventBeginElement name attrs) testType
    | name=="{jabber:client}presence"
    && matchAttribMaybe "type" testType attrs
    = True
isClientPresenceOf _ _ = False

handlePresenceProbe session stanza = do
    withJust (lookupAttrib "to" (tagAttrs stanza)) $ \to -> do
    -- withJust (lookupAttrib "from" (tagAttrs stanza)) $ \from -> do
    jid <- liftIO $ parseAddressJID $ textToByteString to
    withJust (name jid) $ \user -> do
    liftIO $ L.putStrLn $ "RECEIVED PROBE "<++>bshow (peerAddress session,to)
    liftIO $ do
        subs <- getSubscribers (peerSessionFactory session) user
        liftIO $ L.putStrLn $ "subscribers for "<++>bshow user<++>": " <++>bshow subs
        forM_ subs $ \jidstr -> do
            handleIO_ (return ()) $ do
                L.putStrLn $ "parsing " <++>jidstr
                sub <- parseHostNameJID jidstr
                putStrLn $ "comparing " ++show (peer sub , peerAddress session)
                when (peer sub == discardPort (peerAddress session)) $ do
                ps <- userStatus session user
                mapM_ (announcePresence session) ps
    return ()

subscribeToPresence subscribers peer_jid user = do
    pjid <- parseAddressJID peer_jid
    if Set.member pjid subscribers
        then return ()
        else return ()

bare (JID n host _) = JID n host Nothing

presenceErrorRemoteNotFound iqid from to = return
    [ EventBeginElement "{stream:client}presence" 
        ( case iqid of { Nothing -> id; Just iqid -> ( attr "id" iqid :) }
           $ [ attr "from" to
             , attr "type" "error"
             ] )
    , EventBeginElement "{stream:client}error"
        [ attr "type" "modify"]
    , EventBeginElement "{urn:ietf:params:xml:ns:xmpp-stanzas}remote-server-not-found"
        []
    , EventEndElement "{urn:ietf:params:xml:ns:xmpp-stanzas}remote-server-not-found"
    , EventEndElement "{stream:client}error"
    , EventEndElement "{stream:client}presence"
    ]

presenceSubscribed from = return
    [ EventBeginElement "{stream:client}presence" 
        [ attr "from" from
        , attr "type" "subscribed"
        ]
    , EventEndElement   "{stream:client}presence"
    ]

clientRequestsSubscription session cmdChan stanza = do
    liftIO $ do
        putStrLn $ "CLIENT PRESENCE SUBSCRIBE " ++ show stanza
        withJust (lookupAttrib "to" (tagAttrs stanza)) $ \to_str0 -> do
        let to_str = S.takeWhile (/='/') to_str0
            from = lookupAttrib "from" (tagAttrs stanza)
            iqid = lookupAttrib "id" (tagAttrs stanza)
        let handleError e | isDoesNotExistError e = do
                putStrLn $ "remote-server-not-found"
                r <- presenceErrorRemoteNotFound iqid from to_str
                atomically $ writeTChan cmdChan (Send r)
            handleError e = do
                putStrLn $ "ERROR: "++ show e
        handleIO handleError $ do
            let to_str' = textToByteString to_str
            to_jid <- fmap bare $ parseHostNameJID to_str'
            if (is_remote . peer) to_jid 
             then do
                addSolicited session to_str' to_jid
                putStrLn $ "added to solicited: " ++ show to_jid
                -- TODO: notify peer.
             else do
                -- addLocalSubscriber session to_str
                -- self <- getJID session
                r <- presenceSubscribed to_str -- self
                atomically $ writeTChan cmdChan (Send r)
    return ()


stanzaFromTo ::
  JabberPeerSession session =>
  session -> Event -> IO (Maybe (JID, JID))
stanzaFromTo session stanza = 
    let lookup key = fmap textToByteString (lookupAttrib key (tagAttrs stanza))
        parse jidstr =  handleIO_ (return Nothing) (fmap Just $ parseAddressJID jidstr)
    in case liftM2 (,) (lookup "from") (lookup "to") of
        Nothing -> return Nothing
        Just (from,to) ->  do
            mfrom <- parse from
            mto   <- parse to
            case liftM2 (,) mfrom mto of
                Nothing -> return Nothing
                Just (from,to) -> do
                    let fromjid = JID (name from) (peerAddress session) Nothing
                    return $ Just (fromjid,to)

peerRequestsSubsription session stanza = do
    liftIO $ putStrLn $ "PEER PRESENCE SUBSCRIBE " ++ show stanza

    whenJust (liftIO . handleIO (\e -> putStrLn ("peerRequestsSubsription: "++show e) >> return Nothing) 
                $ stanzaFromTo session stanza) $ \(fromjid,tojid) -> do
    withJust (name tojid) $ \user -> do

    subs <- liftIO $ do
             subs <- getSubscribers (peerSessionFactory session) user
             msubs <- flip mapM subs $ \str -> do
                        handleIO_ (return Nothing) 
                                  (fmap Just $ parseHostNameJID str)
             return (catMaybes msubs)
    if elem fromjid subs
      then do
        liftIO . L.putStrLn $ bshow fromjid <++> " already subscribed to " <++> user
        -- if already subscribed, reply
        liftIO $ do
             sendPeerMessage session (Approval tojid fromjid)
             ps <- userStatus session user
             mapM_ (announcePresence session) ps
      else
        liftIO $ processRequest session user fromjid

clientApprovesSubscription session stanza = do
    liftIO $ putStrLn $ "CLIENT APPROVES SUBSCRIPTION"
    withJust (lookupAttrib "to" (tagAttrs stanza)) $ \to_str -> do
    liftIO $ approveSubscriber session (textToByteString to_str)

peerApprovesSubscription session stanza = do
    liftIO $ putStrLn $ "PEER APPROVES SUBSCRIPTION"
    whenJust (liftIO . handleIO (\e -> putStrLn ("peerApprovesSubscription: "++show e) >> return Nothing) 
             $ stanzaFromTo session stanza) $ \(fromjid,tojid) -> do
    withJust (name tojid) $ \user -> do
    liftIO $ processApproval session user fromjid

fromPeer :: (MonadThrow m,MonadIO m, JabberPeerSession session) => 
    session -> Sink XML.Event m ()
fromPeer session = doNestingXML $ do
    let log  = liftIO . L.putStrLn . ("(P) " <++>)
    withXML $ \begindoc -> do
    when (begindoc==EventBeginDocument) $ do
    log "begin-doc"
    withXML $ \xml -> do
    withJust (elementAttrs "stream" xml) $ \stream_attrs -> do
    log $ "stream atributes: " <++> bshow stream_attrs

    fix $ \loop -> do
        log "waiting for stanza."
        whenJust nextElement $ \stanza -> do
            stanza_lvl <- nesting

            let unhandledStanza = do
                    xs <- gatherElement stanza Seq.empty
                    prettyPrint "P: " (toList xs)
            case () of
              _ | stanza `isPresenceOf` presenceTypeOnline
                    -> handlePeerPresence session stanza True
              _ | stanza `isPresenceOf` presenceTypeOffline
                    -> handlePeerPresence session stanza False
              _ | stanza `isPresenceOf` presenceTypeProbe
                    -> handlePresenceProbe session stanza
              _ | stanza `isPresenceOf` presenceTypeSubscribe
                    -> peerRequestsSubsription session stanza
              _ | stanza `isPresenceOf` presenceTypeSubscribed
                    -> peerApprovesSubscription session stanza
              _ -> unhandledStanza

            awaitCloser stanza_lvl
            loop

    log $ "end of stream"
    withXML $ \xml -> do
    log $ "end-of-document: " <++> bshow xml




newServerConnections = newTVar Map.empty

data CachedMessages = CachedMessages 
    { presences :: Map JID JabberShow
    , probes :: Map JID (Set (Bool,JID)) -- False means solicitation rather than probe
    , approvals :: Map JID (Set JID) 
    }
newCache = CachedMessages Map.empty Map.empty Map.empty

connect_to_server chan peer = (>> return ()) . runMaybeT $ do
    let port = 5269 :: Int
    -- We'll cache Presence notifications until the socket
    -- is ready.
    cached <- liftIO $ newIORef newCache

    let mmInsert val key mm = Map.alter f key mm
            where 
                f Nothing    = Just $ Set.singleton val
                f (Just set) = Just $ Set.insert val set
    let cacheCmd (OutBoundPresence (Presence jid Offline)) cached = do
                    cache <- readIORef cached
                    writeIORef cached (cache { presences=Map.delete jid . presences $ cache })
        cacheCmd (OutBoundPresence p@(Presence jid st)) cached = do
                    cache <- readIORef cached
                    writeIORef cached (cache { presences=Map.insert jid st . presences $ cache })
        cacheCmd (PresenceProbe from to) cached = do
                    cache <- readIORef cached
                    let probes' = mmInsert  (True,from) to $ probes cache
                    writeIORef cached (cache { probes=probes' })
        cacheCmd (Solicitation from to) cached = do
                    cache <- readIORef cached
                    let probes' = mmInsert (False,from) to $ probes cache
                    writeIORef cached (cache { probes=probes' })
        cacheCmd (Approval from to) cached = do
                    cache <- readIORef cached
                    let approvals' = mmInsert from to $ approvals cache
                    writeIORef cached (cache { approvals=approvals' })

    fix $ \sendmsgs -> do
        connected <- liftIO . async $ connect' (peerAddr peer) port

        sock <- MaybeT . fix $ \loop -> do
            e <- atomically $ orElse
                    (fmap Right $ waitSTM connected)
                    (fmap Left  $ readTChan chan)
            case e of
                Left cmd   -> cacheCmd cmd cached >> loop
                Right sock -> return sock

        retry <- do
            (cache,snk) <- liftIO $ do
                h <- socketToHandle sock ReadWriteMode
                hSetBuffering h NoBuffering
                cache <- readIORef $ cached
                -- hint garbage collector: we're done with this...
                writeIORef cached newCache
                return (cache,packetSink h)
            MaybeT $ handleOutgoingToPeer (restrictSocket sock) cache chan snk

        liftIO $ cacheCmd retry cached
        liftIO $ putStrLn $ "retrying " ++ show retry
        sendmsgs


greetPeer =
    [ EventBeginDocument
    , EventBeginElement (streamP "stream")
        [ attr "xmlns"   "jabber:server"
        , attr "version" "1.0"
        ]
    ]

goodbyePeer = 
    [ EventEndElement (streamP "stream")
    , EventEndDocument
    ]

presenceStanza sock fromjid tojid typ = do
    addr <- getSocketName sock
    let jidstr jid = toStrict . L.decodeUtf8 
                      $ name jid <$++> "@" 
                        <?++> showPeer (RemotePeer addr) 
                        <++?>  "/" <++$> resource jid
        from = jidstr fromjid
        to = toStrict . L.decodeUtf8
              $ name tojid <$++> "@"
                <?++> showPeer (peer tojid)
    return 
      [ EventBeginElement "{jabber:server}presence"
            [ attr "from" from
            , attr "to"   to
            , attr "type" typ
            ]
      , EventEndElement "{jabber:server}presence"
      ]

{-
toPeerChain
  :: SocketLike sock =>
     sock
     -> CachedMessages
     -> TChan OutBoundMessage
     -> Sink ByteString IO b
     -> IO b
toPeerChain sock cache chan snk = toPeer sock cache chan $$ renderChunks =$ snk
-}

toPeer
  :: SocketLike sock =>
     sock
     -> CachedMessages
     -> TChan OutBoundMessage
     -> (Maybe OutBoundMessage -> IO ())
     -> ConduitM i [Event] IO ()
toPeer sock cache chan fail = do
    let -- log     = liftIO . L.putStrLn . ("(>P) " <++>)
        send xs = yield xs >> prettyPrint ">P: " xs -- >> return (3::Int)
        checkConnection cmd = do
            liftIO $ catch (getPeerName sock >> return ())
                           (\_ -> fail . Just $ cmd)
        sendOrFail getXML cmd = do
            checkConnection cmd
            r <- liftIO $ getXML
                    -- handleIO (\e -> putStrLn ("ERROR: "++show e) >> return []) getXML
            yieldOr r (fail . Just $ cmd)
            prettyPrint ">P: " r
        sendPresence presence =
            sendOrFail (xmlifyPresenceForPeer sock presence)
                       (OutBoundPresence presence)
        sendProbe from to =
            sendOrFail (presenceStanza sock from to "probe")
                       (PresenceProbe from to)
        sendSolicitation from to = 
            sendOrFail (presenceStanza sock from to "subscribe")
                       (Solicitation from to)
        sendApproval from to = 
            sendOrFail (presenceStanza sock from to "subscribed")
                       (Approval from to)


    send greetPeer
    forM_ (Map.assocs . approvals $ cache) $ \(to,froms) -> do
        forM_ (Set.toList froms) $ \from -> do
            liftIO $ L.putStrLn "sending cached approval..."
            sendApproval from to
    forM_ (Map.assocs . presences $ cache) $ \(jid,st) -> do
        sendPresence (Presence jid st)
    forM_ (Map.assocs . probes $ cache) $ \(to,froms) -> do
        forM_ (Set.toList froms) $ \(got,from) -> do
            if got 
              then do
                    liftIO $ L.putStrLn "sending cached probe..."
                    sendProbe from to
              else do
                    liftIO $ L.putStrLn "sending cached solicitation..."
                    sendSolicitation from to

                    
    fix $ \loop -> do
        event <- lift . atomically $ readTChan chan
        case event of
          OutBoundPresence p    -> sendPresence p
          PresenceProbe from to -> do
            liftIO $ L.putStrLn "sending live probe..."
            sendProbe from to
          Solicitation from to -> do
            liftIO $ L.putStrLn "sending live solicitation..."
            sendSolicitation from to
          Approval from to -> do
            liftIO . L.putStrLn $ "sending approval "<++>bshow (from,to)
            sendApproval from to
        loop
    send goodbyePeer

handleOutgoingToPeer
  :: SocketLike sock =>
     sock
     -> CachedMessages
     -> TChan OutBoundMessage
     -> Sink ByteString IO ()
     -> IO (Maybe OutBoundMessage)
handleOutgoingToPeer sock cache chan snk = do
    p <- getPeerName sock
    L.putStrLn $ "(>P) connected " <++> showPeer (RemotePeer p)
    failed <- newIORef Nothing
    let failure cmd = do
            writeIORef failed cmd
            putStrLn $ "Failed: " ++ show cmd
    finally (
#ifdef RENDERFLUSH
        handle (\(IOError _ _ _ _ _ _) -> return ()) $ 
            toPeer sock cache chan failure
                $$  flushList
                =$= renderBuilderFlush def
                =$= builderToByteStringFlush
                =$= discardFlush
                 =$ snk
#else
        handle (\(IOError _ _ _ _ _ _) -> return ()) $ toPeer sock cache chan failure $$ renderChunks =$ snk
#endif
        ) $ L.putStrLn $ "(>P) disconnected " <++> showPeer (RemotePeer p)
    readIORef failed

connect' :: SockAddr -> Int -> IO (Maybe Socket)
connect' addr port = do
    proto <- getProtocolNumber "tcp"
    {-
    -- Given (host :: HostName) ...
    let hints = defaultHints { addrFlags = [AI_ADDRCONFIG]
                             , addrProtocol = proto
                             , addrSocketType = Stream }
    addrs <- getAddrInfo (Just hints) (Just host) (Just serv)
    firstSuccessful $ map tryToConnect addrs
    -}
    let getport (SockAddrInet port _) = port
        getport (SockAddrInet6 port _ _ _) = port
    let doException e@(IOError _ _ _ _ _ _) = do
    -- let doException (SomeException e) = do
            L.putStrLn $  "\nFailed to reach "<++> showPeer (RemotePeer addr) <++> " on port "<++>bshow port<++>": " <++> bshow e
            return Nothing
    handle doException
         $ tryToConnect proto (addr `withPort` port)
  where
  tryToConnect proto addr =
    bracketOnError
        (socket (socketFamily addr) Stream proto)
        (sClose )  -- only done if there's an error
        (\sock -> do
          connect sock addr
          return (Just sock) -- socketToHandle sock ReadWriteMode
        )


sendMessage  :: TVar (Map Peer (TChan OutBoundMessage, ThreadId)) -> OutBoundMessage -> Peer -> IO ()
sendMessage cons msg peer0 = do
    let peer = discardPort peer0
    found <- atomically $ do
        consmap <- readTVar cons
        return (Map.lookup peer consmap)
    let newEntry = do
          chan <- atomically newTChan
          t <- forkIO $ connect_to_server chan peer
          -- L.putStrLn $ "remote-map new: " <++> showPeer peer
          return (True,(chan,t))
    (is_new,entry) <- maybe newEntry
          ( \(chan,t) -> do 
                        st <- threadStatus t
                        let running = do
                                -- L.putStrLn $ "remote-map, thread running: " <++> showPeer peer
                                return (False,(chan,t))
                            died = do
                                -- L.putStrLn $ "remote-map, thread died("<++>bshow st<++>"): " <++> showPeer peer
                                newEntry 
                        case st of 
                          ThreadRunning   -> running
                          ThreadBlocked _ -> running
                          ThreadDied      -> died
                          ThreadFinished  -> died
                        )
          found
    -- L.putStrLn $ "sendMessage ->"<++>showPeer peer<++>": "<++>bshow msg
    atomically $ writeTChan (fst entry) msg
    when is_new . atomically $
        readTVar cons >>= writeTVar cons . Map.insert peer entry



seekRemotePeers :: JabberPeerSession config =>
     XMPPPeerClass config -> TChan Presence -> TVar (Map Peer (TChan OutBoundMessage, ThreadId)) -> IO b0
seekRemotePeers config chan server_connections = do
    fix $ \loop -> do
        event <- atomically $ readTChan chan
        case event of
            p@(Presence jid stat) | not (is_remote (peer jid)) -> do
              -- L.putStrLn $ "seekRemotePeers: " <++> L.show jid <++> " " <++> bshow stat
              runMaybeT $ do
                  u <- MaybeT . return $ name jid
                  subscribers <- liftIO $ do
                    subs <- getSubscribers config u
                    mapM parseHostNameJID subs
                  -- liftIO . L.putStrLn $ "subscribers: " <++> bshow subscribers
                  let peers = Set.map peer (Set.fromList subscribers)
                  forM_ (Set.toList peers) $ \peer -> do
                    when (is_remote peer) $
                        liftIO $ sendMessage server_connections (OutBoundPresence p) peer
                  -- TODO: send presence probes for buddies
                  -- TODO: cache remote presences for clients
            _ -> return (Just ())
        loop 

xmlifyPresenceForPeer sock (Presence jid stat) = do
    -- TODO: accept socket argument and determine local ip address
    --       connected to this peer.
    addr <- getSocketName sock
    let n = name jid
        rsc = resource jid
        jidstr = toStrict . L.decodeUtf8 
                  $ n <$++> "@" <?++> showPeer (RemotePeer addr) <++?>  "/" <++$> rsc 
    return 
      [ EventBeginElement "{jabber:server}presence" 
            (attr "from" jidstr:typ stat)
      , EventBeginElement "{jabber:server}show" []
      , EventContent (ContentText . shw $ stat)
      , EventEndElement "{jabber:server}show"
      , EventEndElement "{jabber:server}presence"
      ]
 where
    typ Offline = [attr "type" "unavailable"]
    typ _       = []
    shw Available = "chat"
    shw Away      = "away"
    shw Offline   = "away" -- Is this right?