{-# LANGUAGE  ExistentialQuantification, DeriveDataTypeable
, FlexibleInstances, FlexibleContexts, UndecidableInstances, MultiParamTypeClasses #-}


module Transient.MapReduce   (Distributable(..),distribute, getText, getUrl, getFile,textUrl, textFile, mapKeyB, mapKeyU, reduce) where
import Transient.Base
import Transient.Internals(onNothing)
import Transient.Move hiding (pack)
import Transient.Logged
import Transient.Indeterminism
import Control.Applicative
import System.Random
import Control.Monad.IO.Class

import System.IO
import Control.Monad
import Data.Monoid
import Data.Maybe

import Data.Typeable
import Data.List hiding (delete, foldl')
import Control.Exception
import Control.Concurrent
--import Data.Time.Clock
import Network.HTTP
import Data.TCache hiding (onNothing)
import Data.TCache.Defs

import Data.ByteString.Lazy.Char8 (pack,unpack)
import Control.Monad.STM
import qualified Data.Map.Strict as M
import Control.Arrow (second)
import qualified Data.Vector.Unboxed as DVU
import qualified Data.Vector as DV
import Data.Hashable
import System.IO.Unsafe

import qualified Data.Foldable as F
import qualified Data.Text as Text

data DDS a= Loggable a => DDS  (Cloud (PartRef a))
data PartRef a= Ref Node Path Save deriving (Typeable, Read, Show)
data Partition a=  Part Node Path Save a deriving (Typeable,Read,Show)
type Save= Bool


instance Indexable (Partition a) where
    key (Part _ string b _)= keyp string b


keyp s True= "PartP@"++s
keyp s False="PartT@"++s

instance Loggable a => IResource (Partition a) where
    keyResource= key
    readResourceByKey k=  r
      where
      typePart :: IO (Maybe a) -> a
      typePart = undefined
      r =  if k !! 4 /= 'P' then return Nothing   else
            defaultReadByKey  (defPath (typePart r) ++ k) >>= return . fmap ( read . unpack)
    writeResource (s@(Part _ _ save _))=
          unless (not save) $ defaultWrite (defPath s ++ key s) (pack $ show s)


eval :: DDS a -> TransIO (PartRef a)
eval (DDS mx) = runCloud mx


type Path=String


instance F.Foldable DVU.Vector where
  {-# INLINE foldr #-}
  foldr = foldr

  {-# INLINE foldl #-}
  foldl = foldl

  {-# INLINE foldr1 #-}
  foldr1 = foldr1

  {-# INLINE foldl1 #-}
  foldl1 = foldl1

--foldlIt' :: V.Unbox a => (b -> a -> b) -> b -> V.Vector a -> b
--foldlIt' f z0 xs= V.foldr f' id xs z0
--      where f' x k z = k $! f z x
--
--foldlIt1 :: V.Unbox a => (a -> a -> a) -> V.Vector a -> a
--foldlIt1 f xs = fromMaybe (error "foldl1: empty structure")
--                    (V.foldl mf Nothing xs)
--      where
--        mf m y = Just (case m of
--                         Nothing -> y
--                         Just x  -> f x y)

class (F.Foldable c, Typeable c, Typeable a, Monoid (c a), Loggable (c a)) => Distributable c a where
   singleton :: a -> c a
   splitAt :: Int -> c a -> (c a, c a)
   fromList :: [a] -> c a

instance (Loggable a) => Distributable DV.Vector a where
   singleton = DV.singleton
   splitAt= DV.splitAt
   fromList = DV.fromList

instance (Loggable a,DVU.Unbox a) => Distributable DVU.Vector a where
   singleton= DVU.singleton
   splitAt= DVU.splitAt
   fromList= DVU.fromList

-- | perform a map and partition the result with different keys using boxed vectors
-- The final result will be used by reduce.
mapKeyB :: (Loggable a, Loggable b,  Loggable k,Ord k)
     => (a -> (k,b))
     -> DDS  (DV.Vector a)
     -> DDS (M.Map k(DV.Vector b))
mapKeyB= mapKey

-- | perform a map and partition the result with different keys using unboxed vectors
-- The final result will be used by reduce.
mapKeyU :: (Loggable a, DVU.Unbox a, Loggable b, DVU.Unbox b,  Loggable k,Ord k)
     => (a -> (k,b))
     -> DDS  (DVU.Vector a)
     -> DDS (M.Map k(DVU.Vector b))
mapKeyU= mapKey

-- | perform a map and partition the result with different keys.
-- The final result will be used by reduce.
mapKey :: (Distributable vector a,Distributable vector b, Loggable k,Ord k)
     => (a -> (k,b))
     -> DDS  (vector a)
     -> DDS (M.Map k (vector b))
mapKey f (DDS mx)= DDS $  do
        refs <-  mx
        process refs

  where
--  process ::  Partition a -> Cloud [Partition b]
  process  (ref@(Ref node path sav))= runAt node $ local $ do
              xs <- getPartitionData ref   -- !> "CMAP"
              generateRef node $ map1 f xs



--  map1 :: (Ord k, F.Foldable vector) => (a -> (k,b)) -> vector a -> M.Map k(vector b)
  map1 f v=  F.foldl' f1  M.empty v
     where
     f1 map x=
           let (k,r) = f x
           in M.insertWith (<>) k (Transient.MapReduce.singleton r) map



--instance Show a => Show (MVar a) where
--   show mvx= "MVar " ++ show (unsafePerformIO $ readMVar mvx)
--
--instance Read a => Read (MVar a) where
--   readsPrec n ('M':'V':'a':'r':' ':s)=
--      let [(x,s')]= readsPrec n s
--      in [(unsafePerformIO $ newMVar x,s')]

data ReduceChunk a= EndReduce | Reduce a deriving (Typeable, Read, Show)

reduce ::  (Hashable k,Ord k, Distributable vector a, Loggable k,Loggable a)
             => (a -> a -> a) -> DDS (M.Map k (vector a)) ->Cloud (M.Map k a)

reduce red  (dds@(DDS mx))= do
   box <- local newMailBox
   nodes <- onAll getNodes

   let lengthNodes = length nodes
       shuffler=   do
          ref <- mx
          local $ do
              mpairs <-  getPartitionData ref   -- !> "GETPARTITIONDATA" ::(M.Map k (vector a))


              runCloud ( shuffle mpairs) <*** runCloud sendEnd

          stop

--       shuffle ::( Hashable k, Distributable vector a) =>M.Map k (vector a) -> Cloud ()
       shuffle mpairs=do
           let destinations= groupByDestiny mpairs  -- ::  M.Map Int (k, vector a)
           parallelize foldAndSend $ M.assocs destinations
           where
--           foldAndSend :: (Hashable k, Distributable vector a)=> (Int,[(k,vector a)]) -> Cloud ()
           foldAndSend (i,kvs)= do
                    let folded= map (\(k,vs) -> (k,foldl1 red vs))  kvs
                    runAt  (nodes !! i) $ local $ putMailBox box $ Reduce folded

            --  agrupa segun nodo de destino:

--           groupByDestiny :: (Hashable k, Distributable vector a)
--                          => M.Map k (vector a)
--                          -> M.Map Int [(k ,vector a)]
           groupByDestiny map= M.foldlWithKey' f M.empty  map
              where
--              f ::  M.Map Int [(k ,vector a)] -> k -> vector a -> M.Map Int [(k ,vector a)]
              f map k vs= M.insertWith f1 (hash1 k) [(k,vs)] map

              f1  v1 v2= v1<> v2
              hash1 k= abs $ hash k `rem` length nodes


       sendEnd =  (clustered $ local $ putMailBox box (EndReduce `asTypeOf` paramOf dds))
                    -- !> "ENDREDUCE"


       reducer=   mclustered reduce1    -- a reduce1 process in each node, get the results and mappend them

--     reduce :: (Ord k)  => Cloud (M.Map k v)

       reduce1 =  local $ do
           reduceResults <- liftIO $ newMVar M.empty  -- !>  "CREATE MVAR for results"
           numberSent <- liftIO $ newMVar 0
           minput <- getMailBox box  -- get the chunk once it arrives to the mailbox

           case minput  of -- !> ("received",minput) of
             EndReduce -> do
                n <- liftIO $ modifyMVar numberSent $ \r -> return (r+1, r+1)
                if n == lengthNodes                         -- !> ( n, lengthNodes)
                 then liftIO $ readMVar reduceResults     -- !> "END reduce"

                 else stop

             Reduce kvs ->


              do
                let addIt (k,inp) = do
                            let input= inp `asTypeOf` atype dds
                            liftIO $ modifyMVar_ reduceResults
                                   $ \map -> do
                                      let maccum =  M.lookup k map
                                      return $ M.insert k (case maccum of
                                        Just accum ->  red input accum
                                        Nothing    ->  input) map
                mapM addIt  (kvs `asTypeOf` paramOf' dds)
                stop


   reducer <|> shuffler
   where
     atype ::DDS(M.Map k (vector a)) ->  a
     atype = undefined -- type level

     paramOf  :: DDS (M.Map k (vector a)) -> ReduceChunk [( k,  a)]
     paramOf = undefined -- type level
     paramOf'  :: DDS (M.Map k (vector a)) ->  [( k,  a)]
     paramOf' = undefined -- type level

{-
reduce red  (dds@(DDS mx))= do
   box <- local newMailBox
   nodes <- onAll getNodes

   let lengthNodes = length nodes
       shuffler=   do
          ref <- mx
          local $ do
              m <-  getPartitionData ref   -- !> "GETPARTITIONDATA"
              let ass= M.assocs m

              runCloud (parallelize shuffle ass) <*** runCloud sendEnd

          stop

--       shuffle ::(k,vector a) -> Cloud ()
       shuffle part | null part = empty
                    | otherwise = do

                         let (k,vs)= part
                             v= foldl1 red vs
                             i=  abs $ hash k `rem` length nodes
                         runAt  (nodes !! i) $ local $ putMailBox box $ Reduce (k,v)
                             -- !> (" PUTMAILBOX ",i,v)
         stop   :: Cloud ()

       sendEnd =  (clustered $ local $ putMailBox box (EndReduce `asTypeOf` paramOf dds))  -- !> "ENDREDUCE"


       reducer=   mclustered reduce    -- a reduce process in each node

--     reduce :: (Ord k)  => Cloud (M.Map k v)

       reduce =  local $ do
           reduceResults <- liftIO $ newMVar M.empty  -- !>  "CREATE MVAR for results"
           numberSent <- liftIO $ newMVar 0
           minput <- getMailBox box  -- get the chunk once it arrives to the mailbox

           case minput  of -- !> ("received",minput) of
             Reduce (k,inp) -> do
                let input= inp `asTypeOf` atype dds

                liftIO $ modifyMVar_ reduceResults
                       $ \map -> do
                          let maccum =  M.lookup k map
                          return $ M.insert k (case maccum of
                            Just accum ->  red input accum
                            Nothing    ->  input) map
                empty

             EndReduce -> do
                n <- liftIO $ modifyMVar numberSent $ \r -> return (r+1, r+1)
                if n == lengthNodes                         -- !> ( n, lengthNodes)
                 then liftIO $ readMVar reduceResults     -- !> "END reduce"

                 else empty

   reducer <|> shuffler
   where
     atype ::DDS(M.Map k (vector a)) ->  a
     atype = undefined -- type level

     paramOf  :: DDS (M.Map k (vector a)) -> ReduceChunk( k,  a)
     paramOf = undefined -- type level
-}


--parallelize :: Loggable b => (a -> Cloud b) -> [a] -> Cloud b
parallelize f xs =  foldr (<|>) empty $ map f xs






getPartitionData :: Loggable a => PartRef a   -> TransIO  a
getPartitionData (Ref node path save)  = do
    (Part _ _ _ xs) <- (liftIO $ atomically
                                   $ readDBRef
                                   $ getDBRef
                                   $ keyp path save)
                              `onNothing` error ("not found DDS data: "++ keyp path save)
    return xs  -- !> "getPartitionData"

-- en caso de fallo de Node, se lanza un clustered en busca del path
--   si solo uno lo tiene, se copia a otro
--   se pone ese nodo de referencia en Part
runAtP :: Loggable a => Node  -> (Path -> IO a) -> Path -> Cloud a
runAtP node f uuid= do
   r <- streamFrom node $ onAll . liftIO $ (SLast <$> f uuid) `catch` sendAnyError
   case r of
     SLast r -> return r
     SError e -> do
         nodes <-  mclustered $ search uuid
         when(length nodes < 1) $ asyncDuplicate node uuid
         runAtP ( head nodes) f uuid

search uuid= error $ "chunk failover not yet defined. Lookin for: "++ uuid

asyncDuplicate node uuid= do
    forkTo node
    nodes <- onAll getNodes
    let node'= head $ nodes \\ [node]
    content <- onAll . liftIO $ readFile uuid
    runAt node' $ local $ liftIO $ writeFile uuid content

sendAnyError :: SomeException -> IO (StreamData a)
sendAnyError e= return $ SError  e


-- | distribute a vector of values among many nodes.
-- If the vector is static and sharable, better use the get* primitives
-- since each node will load the data independently.
distribute :: (Loggable a, Distributable vector a ) => vector a -> DDS (vector a)
distribute = DDS . distribute'

distribute' xs= loggedc $  do
   nodes <- local getNodes                                        -- !> "DISTRIBUTE"
   let lnodes = length nodes
   let size= case F.length xs `div` (length nodes) of 0 ->1 ; n -> n
       xss= split size lnodes 1 xs                                     -- !> size
   distribute'' xss nodes
   where
   split n s s' xs | s==s' = [xs]
   split n s s' xs=
      let (h,t)= Transient.MapReduce.splitAt n xs
      in h : split n s (s'+1) t

distribute'' :: (Loggable a, Distributable vector a) => [vector a] -> [Node] -> Cloud (PartRef (vector a))
distribute'' xss nodes =
   parallelize  move $ zip nodes xss   -- !> show xss
   where
   move (node, xs)=  runAt node $ local $ do
                        par <- generateRef node xs
                        return  par

-- | input data from a text that must be static and shared by all the nodes
getText  :: (Loggable a, Distributable vector a) => (String -> [a]) -> String -> DDS (vector a)
getText part str= DDS $ do
   nodes <- onAll getNodes                                        -- !> "DISTRIBUTE"
   let lnodes = length nodes

   parallelize  (process lnodes)  $ zip nodes [0..lnodes]    -- !> show xss
   where
   process lnodes (node,i)= runAt node $ local $   do
                        let xs = part str
                            size= case length xs `div` lnodes of 0 ->1 ; n -> n
                            xss= Transient.MapReduce.fromList $ take size $ drop  (i *  size) xs
                        par <- generateRef node xss
                        return  par

-- | get the worlds of an URL
textUrl :: String -> DDS (DV.Vector Text.Text)
textUrl= getUrl  (map Text.pack . words)

-- | generate a DDS from the content of a URL
getUrl :: (Loggable a, Distributable vector a) => (String -> [a]) -> String -> DDS (vector a)
getUrl partitioner url= DDS $ do
   nodes <- onAll getNodes                                        -- !> "DISTRIBUTE"
   let lnodes = length nodes

   parallelize  (process lnodes)  $ zip nodes [0..lnodes]    -- !> show xss
   where
   process lnodes (node,i)=  runAt node $ local $ do
                        r <- liftIO . simpleHTTP $ getRequest url
                        body <- liftIO $  getResponseBody r
                        let xs = partitioner body
                            size= case length xs `div` lnodes of 0 ->1 ; n -> n
                            xss= Transient.MapReduce.fromList $ take size $ drop  (i *  size) xs
                        generateRef node xss


-- | get the words of a file
textFile ::  String -> DDS (DV.Vector Text.Text)
textFile= getFile (map Text.pack . words)

-- | generate a DDS from a file. All the nodes must access the file with the same path
-- the first parameter is the parser that generates elements from the content
getFile :: (Loggable a, Distributable vector a) => (String -> [a]) ->  String -> DDS (vector a)
getFile partitioner file= DDS $ do
   nodes <- local getNodes                                        -- !> "DISTRIBUTE"
   let lnodes = length nodes

   parallelize  (process lnodes) $ zip nodes [0..lnodes]    -- !> show xss
   where
   process lnodes (node, i)=  runAt node $ local $ do
                        content <- liftIO $ readFile file
                        let xs = partitioner content
                            size= case length xs `div` lnodes of 0 ->1 ; n -> n
                            xss=Transient.MapReduce.fromList $ take size $ drop  (i *  size) xs                                -- !> size
                        generateRef node   xss



generateRef :: Loggable a => Node -> a -> TransIO (PartRef a)
generateRef node x=  liftIO $ do
       temp <- getTempName
       let reg=  Part node temp True  x
       atomically $ newDBRef reg
       return $ getRef reg

getRef (Part n t s x)= Ref n t s

getTempName :: IO String
getTempName=  ("DDS" ++) <$> replicateM  5 (randomRIO ('a','z'))


-------------- Distributed  Datasource Streams ---------
-- | produce a stream of DDS's that can be map-reduced. Similar to spark streams.
-- each interval of time,a new DDS is produced.(to be tested)
streamDDS
  :: (Loggable a, Distributable vector a) =>
     Integer -> IO (StreamData a) -> DDS (vector a)
streamDDS time io= DDS $ do
     xs <- local . groupByTime time $ do
               r <- parallel io
               case r of
                    SDone -> empty
                    SLast x -> return x
                    SMore x -> return x
                    SError e -> error $ show e
     distribute'  $ Transient.MapReduce.fromList xs




--data CloudArray a= Cloud [a]
--
--instance functor  CloudArray where
--   fmap f mx= do
--        xs <- mx
--        xss <- partition xs
--        rss <- clustered f xss
--        return $ concat rss