summaryrefslogtreecommitdiff
path: root/src/Network/BitTorrent/DHT/Query.hs
blob: 003bb5b934e84efa3fabb4a94ec3acf9856e1c3b (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
-- |
--   Copyright   :  (c) Sam Truzjan 2014
--   License     :  BSD3
--   Maintainer  :  pxqr.sta@gmail.com
--   Stability   :  experimental
--   Portability :  portable
--
--   This module provides functions to interact with other nodes.
--   Normally, you don't need to import this module, use
--   "Network.BitTorrent.DHT" instead.
--
{-# LANGUAGE CPP                 #-}
{-# LANGUAGE FlexibleContexts    #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TemplateHaskell     #-}
{-# LANGUAGE TupleSections       #-}
{-# LANGUAGE PartialTypeSignatures #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE MultiParamTypeClasses #-}
module Network.BitTorrent.DHT.Query
       ( -- * Handler
         -- | To bind specific set of handlers you need to pass
         -- handler list to the 'startNode' function.
         pingH
       , findNodeH
       , getPeersH
       , announceH
       , defaultHandlers

         -- * Query
         -- ** Basic
         -- | A basic query perform a single request expecting a
         -- single response.
       , Iteration
       , pingQ
       , coldPingQ
       , findNodeQ
       , getPeersQ
       , announceQ

         -- ** Iterative
         -- | An iterative query perform multiple basic queries,
         -- concatenate its responses, optionally yielding result and
         -- continue to the next iteration.
       , Search
       -- , search
       , publish
       , ioFindNode
       , ioFindNodes
       , ioGetPeers
       , isearch
       , bgsearch

       -- ** Routing table
       , insertNode
       , refreshNodes

       -- ** Messaging
       , queryNode
       , queryNode'
       , (<@>)
       ) where

import Data.Bits
import Data.Default
#ifdef THREAD_DEBUG
import Control.Concurrent.Lifted.Instrument hiding (yield)
#else
import GHC.Conc (labelThread)
import Control.Concurrent.Lifted hiding (yield)
#endif
import Control.Exception.Lifted hiding (Handler)
import Control.Monad.Reader
import Control.Monad.Logger
import Data.Maybe
import Data.Conduit
import Data.Conduit.List as C hiding (mapMaybe, mapM_)
import Data.Either
import Data.List as L
import Data.Monoid
import Data.Text as T
import qualified Data.Set as Set
         ;import Data.Set (Set)
import Network
import Text.PrettyPrint as PP hiding ((<>), ($$))
import Text.PrettyPrint.HughesPJClass hiding ((<>),($$))
import Data.Time
import Data.Time.Clock.POSIX
import Data.Hashable (Hashable)
import Data.Serialize
import Data.Hashable

import Network.DatagramServer as KRPC hiding (Options, def)
import Network.KRPC.Method    as KRPC
import Network.DatagramServer.Mainline (ReflectedIP(..), QueryExtra(..), ResponseExtra(..))
import Network.DatagramServer (QueryFailure(..))
import Data.Torrent
import qualified Network.DHT as DHT
import Network.DHT.Mainline
import Network.DHT.Routing as R
import Network.BitTorrent.DHT.Session
import Control.Concurrent.STM
import qualified Network.BitTorrent.DHT.Search as Search
#ifdef VERSION_bencoding
import Data.BEncode (BValue)
import Network.DatagramServer.Mainline (KMessageOf)
#else
import Data.ByteString (ByteString)
import Network.DatagramServer.Tox
#endif
import Network.Address hiding (NodeId)
import Network.DatagramServer.Types as RPC hiding (Query,Response)
import Network.DHT.Types
import Control.Monad.Trans.Control
import Data.Typeable
import Data.Serialize
import System.IO.Unsafe (unsafeInterleaveIO)
import Data.String


{-----------------------------------------------------------------------
--  Handlers
-----------------------------------------------------------------------}

{-
nodeHandler :: ( Address ip
               , KRPC dht (Query KMessageOf a) (Response KMessageOf b)
               )
           => (NodeInfo KMessageOf ip () -> Maybe ReflectedIP -> IO ()) -> (NodeAddr ip -> IO (NodeId KMessageOf)) -> (Char -> String -> Text -> IO ()) -> QueryMethod KMessageOf -> (NodeAddr ip -> a -> IO b) -> NodeHandler
-}
nodeHandler :: forall raw dht addr u t q r.
    (Address addr, WireFormat raw dht, Pretty (NodeInfo dht addr u),
     Default u,
     IsString t, Functor dht,
     KRPC dht (Query dht q) (Response dht r),
     SerializableTo raw (Response dht r),
     SerializableTo raw (Query dht q),
     Show (QueryMethod dht)) =>
    (NodeInfo dht addr u -> Maybe ReflectedIP -> IO ())
    -> (NodeAddr addr -> IO (NodeId dht))
    -> (Char -> t -> Text -> IO ())
    -> DHTData dht addr
    -> QueryMethod dht
    -> (NodeAddr addr -> q -> IO r)
    -> Handler IO dht raw
nodeHandler insertNode myNodeIdAccordingTo logm dta method action = handler (\sockaddr -> myNodeIdAccordingTo (error "todo")) method $ \ sockAddr msg -> do
  let remoteId = messageSender (msg :: dht (Query dht q)) resptype
      qextra   = queryExtra qry
      resptype = Proxy :: Proxy (Response dht r)
      q        = queryParams qry
      qry      = envelopePayload msg :: Query dht q
  case fromSockAddr sockAddr of
    Nothing    -> throwIO BadAddress
    Just naddr -> do
      logm 'D' "nodeHandler" $ "Received query: " <> T.pack (show $ method)
      me <- myNodeIdAccordingTo naddr
      rextra <- liftIO $ makeResponseExtra dta me qry resptype
      let ni = NodeInfo remoteId naddr def
      -- Do not route read-only nodes. (bep 43)
      if fromRoutableNode qextra
        then insertNode ni Nothing >> return () -- TODO need to block. why?
        else logm 'W' "nodeHandler" $ "READ-ONLY " <> T.pack (show $ pPrint ni)
      Response
        <$> pure rextra
        <*> action naddr q

-- | Default 'Ping' handler.
pingH :: DHT.Kademlia dht => Proxy dht -> NodeAddr ip -> Ping dht -> IO (Ping dht)
pingH dht _ _ = return (DHT.pongMessage dht)
-- pingH = nodeHandler $ \ _ p@PingPayload{} -> return p { isPong = True }

-- | Default 'FindNode' handler.
findNodeH :: Kademlia dht => (NodeId dht -> IO [NodeInfo dht ip u]) -> NodeAddr ip -> FindNode dht ip -> IO (NodeFound dht ip)
findNodeH getclosest _ msg = foundNodesMessage . L.map (fmap (const ())) <$> getclosest (findWho msg)

-- | Default 'GetPeers' handler.
getPeersH :: Hashable ip => (InfoHash -> IO (PeerList ip)) -> TVar SessionTokens -> NodeAddr ip -> GetPeers ip -> IO (GotPeers ip)
getPeersH getPeerList toks naddr (GetPeers ih) = do
  ps <- getPeerList ih
  tok <- grantToken toks naddr
  return $ GotPeers ps tok

-- | Default 'Announce' handler.
announceH :: ( Ord ip, Hashable ip ) => TVar (PeerStore ip) -> TVar SessionTokens -> NodeAddr ip -> Announce -> IO Announced
announceH peers toks naddr @ NodeAddr {..} (Announce {..}) = do
  valid <- checkToken toks naddr sessionToken
  unless valid $ do
    throwIO $ InvalidParameter "token"

  let annPort  = if impliedPort then nodePort else port
      peerAddr = PeerAddr Nothing nodeHost annPort
  insertPeer peers topic announcedName peerAddr
  return Announced

-- | Includes all Kademlia-related handlers.
kademliaHandlers :: forall raw dht u ip. (Eq ip, Ord ip, Address ip
                    , Ord (TransactionID dht)
                    , Ord (NodeId dht)
                    , Show u
                    , SerializableTo raw (Response dht (Ping dht))
                    , SerializableTo raw (Query dht (Ping dht))
                    , Show (QueryMethod dht)
                    , Show (NodeId dht)
                    , FiniteBits (NodeId dht)
                    , Default u
                    , Serialize (TransactionID dht)
                    , WireFormat raw dht
                    , Kademlia dht
                    , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
                    , Functor dht
                    , Pretty (NodeInfo dht ip u)
                    , KRPC dht (Query dht (FindNode dht ip)) (Response dht (NodeFound dht ip))
                    , SerializableTo raw (Response dht (NodeFound dht ip))
                    , SerializableTo raw (Query dht (FindNode dht ip))
                    ) => LogFun -> DHT raw dht u ip [Handler IO dht raw]
-- kademliaHandlers :: forall ip. (Eq ip, Ord ip, Address ip) => LogFun -> DHT BValue KMessageOf () ip [NodeHandler]
kademliaHandlers logger = do
    groknode <- insertNode1
    mynid <- myNodeIdAccordingTo1
    dta <- asks dhtData
    let handler :: ( KRPC dht (Query dht a) (Response dht b)
                   , SerializableTo raw (Response dht b)
                   , SerializableTo raw (Query dht a)
                   ) => QueryMethod dht -> (NodeAddr ip -> a -> IO b) -> Handler IO dht raw
        handler = nodeHandler groknode mynid (logt logger) dta
        dht = Proxy :: Proxy dht
    getclosest <- getClosest1
    return [ handler (namePing dht)      $ pingH dht
           , handler (nameFindNodes dht) $ findNodeH getclosest
           ]

instance DataHandlers BValue KMessageOf where
    dataHandlers = bthandlers

bthandlers ::
    ( Ord ip , Hashable ip, Typeable ip, Serialize ip) =>
     (NodeId KMessageOf  -> IO [NodeInfo KMessageOf ip ()])
    -> DHTData KMessageOf ip
    -> [MethodHandler BValue KMessageOf ip]
bthandlers getclosest dta =
    [ MethodHandler "get_peers" $ getPeersH (getpeers dta) (sessionTokens dta)
    , MethodHandler "announce_peer" $ announceH (contactInfo dta) (sessionTokens dta)
    ]
 where
    getpeers dta ih = do
      ps <- lookupPeers (contactInfo dta) ih
      if L.null ps
        then Left <$> getclosest (toNodeId ih)
        else return (Right ps)


-- | Includes all default query handlers.
defaultHandlers :: forall raw dht u ip.
                    ( Ord (TransactionID dht)
                    , Ord (NodeId dht)
                    , Show u
                    , SerializableTo raw (Response dht (Ping dht))
                    , SerializableTo raw (Query dht (Ping dht))
                    , Show (QueryMethod dht)
                    , Show (NodeId dht)
                    , FiniteBits (NodeId dht)
                    , Default u
                    , Serialize (TransactionID dht)
                    , WireFormat raw dht
                    , Kademlia dht
                    , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
                    , Functor dht
                    , Pretty (NodeInfo dht ip u)
                    , KRPC dht (Query dht (FindNode dht ip)) (Response dht (NodeFound dht ip))
                    , SerializableTo raw (Response dht (NodeFound dht ip))
                    , SerializableTo raw (Query dht (FindNode dht ip))
                    , Eq ip, Ord ip, Address ip, DataHandlers raw dht
                    ) => LogFun -> DHT raw dht u ip [Handler IO dht raw]
defaultHandlers logger = do
    groknode <- insertNode1
    mynid <- myNodeIdAccordingTo1
    dta <- asks dhtData
    let handler :: MethodHandler raw dht ip -> Handler IO dht raw
        handler (MethodHandler name action) = nodeHandler groknode mynid (logt logger) dta name action
    getclosest <- getClosest1
    hs <- kademliaHandlers logger
    return $ hs ++ L.map handler (dataHandlers (fmap (fmap (fmap (const ()))) . getclosest) dta)

{-----------------------------------------------------------------------
--  Basic queries
-----------------------------------------------------------------------}

type Iteration raw dht u ip o = NodeInfo dht ip u -> DHT raw dht u ip (Either [NodeInfo dht ip u] [o ip])

-- | The most basic query. May be used to check if the given node is
-- alive or get its 'NodeId'.
pingQ :: forall raw dht u ip.
            ( DHT.Kademlia dht
            , Address ip
            , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
            , Default u
            , Show u
            , Ord (TransactionID dht)
            , Serialize (TransactionID dht)
            , WireFormat raw dht
            , SerializableTo raw (Response dht (Ping dht))
            , SerializableTo raw (Query dht (Ping dht))
            , Ord (NodeId dht)
            , FiniteBits (NodeId dht)
            , Show (NodeId dht)
            , Show (QueryMethod dht)
            ) => NodeInfo dht ip u -> DHT raw dht u ip (NodeInfo dht ip u , Maybe ReflectedIP)
pingQ ni = do
  let ping = DHT.pingMessage (Proxy :: Proxy dht)
  (nid, pong, mip) <- queryNode' ni ping
  let _ = pong `asTypeOf` ping
  -- (nid, PingPayload{}, mip) <- queryNode' addr PingPayload {isPong=False, pingId=pid}
  return (NodeInfo nid (nodeAddr ni) def, mip)

-- | The most basic query. May be used to check if the given node is
-- alive or get its 'NodeId'.
coldPingQ :: forall raw dht u ip.
            ( DHT.Kademlia dht
            , Address ip
            , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
            , Default u
            , Show u
            , Ord (TransactionID dht)
            , Serialize (TransactionID dht)
            , WireFormat raw dht
            , SerializableTo raw (Response dht (Ping dht))
            , SerializableTo raw (Query dht (Ping dht))
            , Ord (NodeId dht)
            , FiniteBits (NodeId dht)
            , Show (NodeId dht)
            , Show (QueryMethod dht)
            ) => PacketDestination dht -> DHT raw dht u ip (NodeInfo dht ip u , Maybe ReflectedIP)
coldPingQ dest = do
  let ping = DHT.pingMessage (Proxy :: Proxy dht)
  naddr <- maybe (throwIO $ QueryFailed ProtocolError "unable to construct NodeAddr from PacketDestination")
                 return
                 $ fromAddr dest
  (nid, pong, mip) <- coldQueryNode' naddr dest ping
  let _ = pong `asTypeOf` ping
  -- (nid, PingPayload{}, mip) <- queryNode' addr PingPayload {isPong=False, pingId=pid}
  return (NodeInfo nid naddr def, mip)

-- TODO [robustness] match range of returned node ids with the
-- expected range and either filter bad nodes or discard response at
-- all throwing an exception
-- findNodeQ :: Address ip => TableKey key => key -> IterationI ip NodeInfo
findNodeQ proxy key ni = do
  closest <- fmap DHT.foundNodes $  DHT.findNodeMessage proxy key <@> ni
  $(logInfoS) "findNodeQ" $ "NodeFound\n"
    <> T.pack (L.unlines $ L.map ((' ' :) . show . pPrint) closest)
  return $ Right closest

#ifdef VERSION_bencoding
getPeersQ :: Address ip => InfoHash -> Iteration BValue KMessageOf () ip PeerAddr
getPeersQ topic ni = do
  GotPeers {..} <- GetPeers topic <@> ni
  let dist = distance (toNodeId topic) (nodeId ni)
  $(logInfoS) "getPeersQ" $ T.pack
         $ "distance: " <> render (pPrint dist) <> " , result: "
        <> case peers of { Left _ -> "NODES"; Right _ -> "PEERS" }
  return peers

announceQ :: Address ip => InfoHash -> PortNumber -> Iteration BValue KMessageOf () ip NodeAddr
announceQ ih p ni = do
  GotPeers {..} <- GetPeers ih <@> ni
  case peers of
    Left  ns
      | False     -> undefined -- TODO check if we can announce
      | otherwise -> return (Left ns)
    Right _ -> do -- TODO *probably* add to peer cache
      Announced <- Announce False ih Nothing p grantedToken <@> ni
      return (Right [nodeAddr ni])
#endif

{-----------------------------------------------------------------------
--  Iterative queries
-----------------------------------------------------------------------}


ioGetPeers :: Address ip => InfoHash -> DHT BValue KMessageOf () ip (NodeInfo KMessageOf ip () -> IO ([NodeInfo KMessageOf ip ()], [PeerAddr ip]))
ioGetPeers ih = do
    session <- ask
    return $ \ni -> runDHT session $ do
        r <- try $ getPeersQ ih ni
        case r of
            Right e -> return $ either (,[]) ([],) e
            Left e -> let _ = e :: QueryFailure in return ([],[])

ioFindNode :: ( DHT.Kademlia dht
              , WireFormat raw dht
              , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
              , Address ip
              , Default u
              , Show u
              , Show (QueryMethod dht)
              , TableKey dht infohash
              , Eq (NodeId dht)
              , Ord (NodeId dht)
              , FiniteBits (NodeId dht)
              , Show (NodeId dht)
              , KRPC dht (Query dht (FindNode dht ip)) (Response dht (NodeFound dht ip))
              , Ord (TransactionID dht)
              , Serialize (TransactionID dht)
              , SerializableTo raw (Response dht (NodeFound dht ip))
              , SerializableTo raw (Query dht (FindNode dht ip))
              , SerializableTo raw (Response dht (Ping dht))
              , SerializableTo raw (Query dht (Ping dht))
              ) => infohash -> DHT raw dht u ip (NodeInfo dht ip u -> IO ([NodeInfo dht ip u], [NodeInfo dht ip u]))
ioFindNode ih = do
    session <- ask
    return $ \ni -> runDHT session $ do
        ns <- fmap DHT.foundNodes $ DHT.findNodeMessage Proxy ih <@> ni
        let ns' = L.map (fmap (const def)) ns
        return $ L.partition (\n -> nodeId n /= toNodeId ih) ns'


-- | Like ioFindNode, but considers all found nodes to be 'Right' results.
ioFindNodes :: ( DHT.Kademlia dht
              , WireFormat raw dht
              , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
              , Address ip
              , Default u
              , Show u
              , Show (QueryMethod dht)
              , TableKey dht infohash
              , Eq (NodeId dht)
              , Ord (NodeId dht)
              , FiniteBits (NodeId dht)
              , Show (NodeId dht)
              , KRPC dht (Query dht (FindNode dht ip)) (Response dht (NodeFound dht ip))
              , Ord (TransactionID dht)
              , Serialize (TransactionID dht)
              , SerializableTo raw (Response dht (NodeFound dht ip))
              , SerializableTo raw (Query dht (FindNode dht ip))
              , SerializableTo raw (Response dht (Ping dht))
              , SerializableTo raw (Query dht (Ping dht))
              ) => infohash -> DHT raw dht u ip (NodeInfo dht ip u -> IO ([NodeInfo dht ip u], [NodeInfo dht ip u]))
ioFindNodes ih = do
    session <- ask
    return $ \ni -> runDHT session $ do
        ns <- fmap DHT.foundNodes $ DHT.findNodeMessage Proxy ih <@> ni
        let ns' = L.map (fmap (const def)) ns
        return ([], ns')

isearch :: ( Ord r
           , Ord ip
           , Ord (NodeId dht)
           , FiniteBits (NodeId dht)
           , TableKey dht ih
           , Show ih) =>
            (ih -> DHT raw dht u ip (NodeInfo dht ip u -> IO ([NodeInfo dht ip u], [r])))
            -> ih
            -> DHT raw dht u ip (ThreadId, Search.IterativeSearch dht u ip r)
isearch f ih = do
    qry <- f ih
    ns <- kclosest 8 ih <$> getTable
    liftIO $ do s <- Search.newSearch qry (toNodeId ih) ns
                a <- fork $ do
                    tid <- myThreadId
                    labelThread tid ("search."++show ih)
                    Search.search s
                    -- atomically \$ readTVar (Search.searchResults s)
                return (a, s)

-- | Background search: fill a lazy list using a background thread.
bgsearch f ih = do
    (tid, s) <- isearch f ih
    let again shown = do
            (chk,fin) <- atomically $ do
                r <- (Set.\\ shown) <$> readTVar (Search.searchResults s)
                if not $ Set.null r
                    then (,) r <$> Search.searchIsFinished s
                    else Search.searchIsFinished s >>= check >> return (Set.empty,True)
            let ps = Set.toList chk
            if fin then return ps
                   else do
                        xs <- unsafeInterleaveIO $ again (shown `Set.union` chk)
                        return $ ps ++ xs
    liftIO $ again Set.empty

type Search raw dht u ip o = Conduit [NodeInfo dht ip u] (DHT raw dht u ip) [o dht ip u]

#if 0

-- TODO: use reorder and filter (Traversal option) leftovers
-- search :: k -> IterationI ip o -> Search ip o
search _ action = do
  awaitForever $ \ batch -> unless (L.null batch) $ do
    $(logWarnS) "search" "start query"
    responses <- lift $ queryParallel (action <$> batch)
    let (nodes, results) = partitionEithers responses
    $(logWarnS) "search" ("done query more:" <> T.pack (show (L.length nodes, L.length results)))
    leftover $ L.concat nodes
    let r = mapM_ yield results
        _ = (action,r) :: (a -> DHT raw dht u ip (Either [a] o), ConduitM [a] o (DHT raw dht u ip) ())
    r

#endif

publish = error "todo"
-- publish :: Address ip => InfoHash -> PortNumber -> DHT BValue KMessageOf () ip ()
-- publish ih p = do
  -- nodes <- getClosest ih
  -- r     <- asks (optReplication . options)
  -- _ <- sourceList [nodes] $= search ih (announceQ ih p) $$ C.take r
  -- return ()


probeNode :: ( Default u
            , Show u
            , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
            , DHT.Kademlia dht
            , Address ip
            , Ord (TransactionID dht)
            , Serialize (TransactionID dht)
            , WireFormat raw dht
            , SerializableTo raw (Response dht (Ping dht))
            , SerializableTo raw (Query dht (Ping dht))
            , Ord (NodeId dht)
            , FiniteBits (NodeId dht)
            , Show (NodeId dht)
            , Show (QueryMethod dht)
            ) => NodeInfo dht ip u -> DHT raw dht u ip (Bool , Maybe ReflectedIP)
probeNode addr = do
  $(logDebugS) "routing.questionable_node" (T.pack (render (pPrint $ nodeAddr addr)))
  result <- try $ pingQ addr
  let _ = fmap (const ()) result :: Either QueryFailure ()
  return $ either (const (False,Nothing)) (\(_,mip)->(True,mip)) result


refreshNodes :: forall raw dht u ip.
                ( Address ip
                , Ord (NodeId dht)
                , Default u
                , FiniteBits (NodeId dht)
                , Pretty (NodeId dht)
                , DHT.Kademlia dht
                , Ord ip
                , Ord (TransactionID dht)
                , SerializableTo raw (Response dht (NodeFound dht ip))
                , SerializableTo raw (Query dht (FindNode dht ip))
                , SerializableTo raw (Response dht (Ping dht))
                , SerializableTo raw (Query dht (Ping dht))
                , Pretty (NodeInfo dht ip u)
                , Show (NodeId dht)
                , Show u
                , Show (QueryMethod dht)
                , Serialize (TransactionID dht)
                , WireFormat raw dht
                , KRPC dht (Query dht (FindNode dht ip)) (Response dht (NodeFound dht ip))
                , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
                )  => NodeId dht -> DHT raw dht u ip () -- [NodeInfo KMessageOf ip ()]
-- FIXME do not use getClosest sinse we should /refresh/ them
refreshNodes nid = do
  $(logDebugS) "routing.refresh_bucket" (T.pack (render (pPrint nid)))
  nodes <- getClosest nid
  do
    -- forM (L.take 1 nodes) \$ \ addr -> do
    -- NodeFound ns <- FindNode nid <@> addr
    -- Expected type: ConduitM [NodeAddr ip] [NodeInfo KMessageOf ip ()] (DHT ip) ()
    --   Actual type: ConduitM [NodeInfo KMessageOf ip ()] [NodeInfo KMessageOf ip ()] (DHT ip) ()
    -- nss <- sourceList [[addr]] \$= search nid (findNodeQ nid) $$ C.consume
    -- nss <- sourceList [nodes] \$= search nid (findNodeQ (Proxy :: Proxy dht) nid) $$ C.consume
    ns <- bgsearch ioFindNodes nid
    $(logWarnS) "refreshNodes" $ "received " <> T.pack (show (L.length ns)) <> " nodes."
    _ <- queryParallel $ flip L.map ns $ \n -> do
        $(logWarnS) "refreshNodes" $ "received node: " <> T.pack (show (pPrint n))
        pingQ n
        -- pingQ takes care of inserting the node.
    return ()
  return () -- \$ L.concat nss

logc :: Char -> String -> DHT raw dht u ip ()
logc 'D' = $(logDebugS) "insertNode" . T.pack
logc 'W' = $(logWarnS)  "insertNode" . T.pack
logc 'I' = $(logInfoS)  "insertNode" . T.pack
logc c   = $(logInfoS)  "insertNode" . T.pack . (c :) . (':' :)

-- | This operation do not block but acquire exclusive access to
--   routing table.
insertNode :: forall raw dht u ip.
              ( Address ip
              , Ord (NodeId dht)
              , FiniteBits (NodeId dht)
              , Show (NodeId dht)
              , Default u
              , Show u
              , DHT.Kademlia dht
              , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
              , Ord (TransactionID dht)
              , WireFormat raw dht
              , Serialize (TransactionID dht)
              , SerializableTo raw (Response dht (Ping dht))
              , SerializableTo raw (Query dht (Ping dht))
              , Ord (NodeId dht)
              , Show (NodeId dht)
              , Show (QueryMethod dht)
              ) => NodeInfo dht ip u -> Maybe ReflectedIP -> DHT raw dht u ip ()
insertNode info witnessed_ip0 = do
    f <- insertNode1
    liftIO $ f info witnessed_ip0

insertNode1 :: forall raw dht u ip.
               ( Address ip
               , Default u
               , Show u
               , Ord (NodeId dht)
               , FiniteBits (NodeId dht)
               , Show (NodeId dht)
               , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
               , DHT.Kademlia dht
               , Ord (TransactionID dht)
               , WireFormat raw dht
               , Serialize (TransactionID dht)
               , SerializableTo raw (Response dht (Ping dht))
               , SerializableTo raw (Query dht (Ping dht))
               , Ord (NodeId dht)
               , Show (NodeId dht)
               , Show (QueryMethod dht)
               ) => DHT raw dht u ip (NodeInfo dht ip u -> Maybe ReflectedIP -> IO ())
insertNode1 = do
  bc <- optBucketCount <$> asks options
  nid <- asks tentativeNodeId
  logm0 <- embed_ (uncurry logc)
  let logm c = logm0 . (c,)
  dht_node_state <- ask -- XXX: This prevents ping probe from modifying the Node state.
  probe0 <- embed probeNode -- probeNode :: Address ip => NodeAddr ip -> DHT ip (Bool, Maybe ReflectedIP)
  let probe n = probe0 n >>= runDHT dht_node_state . restoreM
      {-
      changeip ip0 arrival = fromMaybe (DHT.fallbackID params) $ do -- warning: recursive
                    ip <- fromSockAddr ip0 :: Maybe ip
                    listToMaybe
                        $ rank id (nodeId $ foreignNode arrival)
                        $ bep42s ip (DHT.fallbackID params) -- warning: recursive
      -}
      params = DHT.TableParameters
        { maxBuckets = bc :: Int
        , fallbackID = nid :: NodeId dht
        , adjustID = dhtAdjustID Proxy (DHT.fallbackID params) :: SockAddr -> Event dht ip u -> NodeId dht
        , logMessage = logm :: Char -> String -> IO ()
        , pingProbe = probe :: NodeInfo dht ip u -> IO (Bool, Maybe ReflectedIP)
        }
  tbl <- asks routingInfo
  let state = DHT.TableKeeper
        { routingInfo = tbl
        , grokNode    = DHT.insertNode params state
        , grokAddress = \_ _ -> return () -- :: Maybe SockAddr -> ReflectedIP -> IO ()
        }
  return $ \info witnessed_ip0 -> DHT.insertNode params state info witnessed_ip0

-- | Throws exception if node is not responding.
queryNode :: forall raw dht u a b ip.
    ( Address ip
    , KRPC dht (Query dht a) (Response dht b)
    , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
    , Default u
    , Show u
    , DHT.Kademlia dht
    , Ord (TransactionID dht)
    , Serialize (TransactionID dht)
    , WireFormat raw dht
    , SerializableTo raw (Response dht b)
    , SerializableTo raw (Query dht a)
    , Ord (NodeId dht)
    , FiniteBits (NodeId dht)
    , Show (NodeId dht)
    , Show (QueryMethod dht)
    , SerializableTo raw (Response dht (Ping dht))
    , SerializableTo raw (Query dht (Ping dht))
    ) => NodeInfo dht ip u -> a -> DHT raw dht u ip (NodeId dht, b)
queryNode addr q = fmap (\(n,b,_) -> (n,b)) $ queryNode' addr q

queryNode' :: forall raw dht u a b ip.
    ( Address ip
    , Default u
    , Show u
    , DHT.Kademlia dht
    , KRPC dht (Query dht a) (Response dht b)
    , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
    , Ord (TransactionID dht)
    , Serialize (TransactionID dht)
    , WireFormat raw dht
    , SerializableTo raw (Response dht b)
    , SerializableTo raw (Query dht a)
    , Ord (NodeId dht)
    , FiniteBits (NodeId dht)
    , Show (NodeId dht)
    , Show (QueryMethod dht)
    , SerializableTo raw (Response dht (Ping dht))
    , SerializableTo raw (Query dht (Ping dht))
    ) => NodeInfo dht ip u -> a -> DHT raw dht u ip (NodeId dht, b, Maybe ReflectedIP)
queryNode' ni q = do
  let addr = nodeAddr ni
      dest = makeAddress (Left $ nodeId ni) (toSockAddr addr)
  coldQueryNode' addr dest q

coldQueryNode' :: forall raw dht u a b ip.
    ( Address ip
    , Default u
    , Show u
    , DHT.Kademlia dht
    , KRPC dht (Query dht a) (Response dht b)
    , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
    , Ord (TransactionID dht)
    , Serialize (TransactionID dht)
    , WireFormat raw dht
    , SerializableTo raw (Response dht b)
    , SerializableTo raw (Query dht a)
    , Ord (NodeId dht)
    , FiniteBits (NodeId dht)
    , Show (NodeId dht)
    , Show (QueryMethod dht)
    , SerializableTo raw (Response dht (Ping dht))
    , SerializableTo raw (Query dht (Ping dht))
    ) => NodeAddr ip -> PacketDestination dht -> a -> DHT raw dht u ip (NodeId dht, b, Maybe ReflectedIP)
coldQueryNode' addr dest q = do
  nid <- myNodeIdAccordingTo $ fromMaybe (error "TODO: coldQueryNode' myNodeIdAccordingTo") $ fromAddr dest
  dta <- asks dhtData
  qextra <- liftIO $ makeQueryExtra dta nid (Proxy :: Proxy (Query dht q)) (Proxy :: Proxy (Response dht b))
  let read_only = False -- TODO: check for NAT issues. (BEP 43)
  -- let KRPC.Method name = KRPC.method :: KRPC.Method dht (Query dht a) (Response dht b)
  mgr <- asks manager
  (Response rextra r, remoteId, witnessed_ip) <- liftIO $ query' mgr dest (Query qextra q)
  -- \$(logDebugS) "queryNode" $ "Witnessed address: " <> T.pack (show witnessed_ip)
  --                             <> " by " <> T.pack (show (toSockAddr addr))
  _ <- insertNode (NodeInfo remoteId addr def) witnessed_ip
  return (remoteId, r, witnessed_ip)

-- | Infix version of 'queryNode' function.
(<@>) :: ( Address ip
         , KRPC dht (Query dht a) (Response dht b)
         , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
         , Default u
         , Show u
         , Show (QueryMethod dht)
         , Ord (NodeId dht)
         , FiniteBits (NodeId dht)
         , Show (NodeId dht)
         , Ord (TransactionID dht)
         , Serialize (TransactionID dht)
         , SerializableTo raw (Response dht b)
         , SerializableTo raw (Query dht a)
         , SerializableTo raw (Response dht (Ping dht))
         , SerializableTo raw (Query dht (Ping dht))
         , WireFormat raw dht
         , Kademlia dht
         ) => a -> NodeInfo dht ip u -> DHT raw dht u ip b
q <@> addr = snd <$> queryNode addr q
{-# INLINE (<@>) #-}