@@ -1,12 +1,5 @@
11{-# LANGUAGE DeriveGeneric #-}
2−{-# LANGUAGE GADTs #-}
32{-# LANGUAGE OverloadedStrings #-}
4−{-# LANGUAGE FlexibleContexts #-}
5−{-# LANGUAGE FlexibleInstances #-}
6−{-# LANGUAGE TypeFamilies #-}
7−{-# LANGUAGE TypeApplications #-}
8−{-# LANGUAGE MultiParamTypeClasses #-}
9−
103
114module Foo.Data
125 (
@@ -16,6 +9,7 @@ module Foo.Data
169 createApp,
1710 createConfig,
1811 User(..),
12+ UserName(..),
1913 Config,
2014 updates,
2115 allUsers,
@@ -23,16 +17,12 @@ module Foo.Data
2317 UserId,
2418 PostId,
2519 Post(..),
26− postId,
2720 allPosts,
2821 addConnection,
2922 removeConnection,
3023 TokenizedResponse(..),
3124 Update(..),
32− getToken,
33− Token,
34− subscribeWithToken,
35− postUpdates
25+ getToken
3626 ) where
3727
3828import GHC.Conc
@@ -60,14 +50,6 @@ import qualified Data.CyclicBuffer as CB
6050import Data.Set ((\\), union)
6151import Data.Monoid
6252import Debug.Trace
63−import Database.Beam
64−import qualified Database.Beam.Postgres as P
65−import Database.Beam.Query
66−import Database.PostgreSQL.Simple (withSavepoint)
67−import Database.Beam.Backend.SQL.BeamExtensions
68−import System.Random.Shuffle
69−import Network.SocketIO
70−import Network.EngineIO (SocketId)
7153
7254type App = ReaderT Config Handler
7355
@@ -75,35 +57,35 @@ class HasConfig c where
7557 getCfg :: c -> Config
7658
7759allUsers :: App [User]
78−allUsers = asks dbConn >>= \conn -> liftIO . P.runBeamPostgres conn . runSelectReturningList . select . all_ . users $ socialDb
60+allUsers = ask >>= liftIO . fmap M.elems . readTVarIO . users
7961
8062post :: PostId -> App Post
8163post pId = do
82− conn <- asks dbConn
83− u <- liftIO . P.runBeamPostgres conn . runSelectReturningOne . flip lookup_ (PostId pId) . posts $ socialDb
64+ cfg <- ask
65+ u <- liftIO . atomically . fmap (M.lookup pId) . readTVar . posts $ cfg
8466 case u of
8567 Just u -> return u
8668 Nothing -> throwError err404
8769
8870user :: UserId -> App User
8971user uId = do
90− conn <- asks dbConn
91− u <- liftIO . P.runBeamPostgres conn . runSelectReturningOne . flip lookup_ (UserId uId) . users $ socialDb
72+ cfg <- ask
73+ u <- liftIO . atomically . fmap (M.lookup uId) . readTVar . users $ cfg
9274 case u of
9375 Just u -> return u
9476 Nothing -> throwError err404
9577
96−allPosts :: App [Post]
97−allPosts = asks dbConn >>= \conn -> liftIO . P.runBeamPostgres conn . runSelectReturningList . select . all_ . posts $ socialDb
78+allPosts :: App (STM [Post])
79+allPosts = asks $ fmap M.elems . readTVar . posts
9880
9981getToken :: App (STM Token)
10082getToken = asks $ fmap CB.lastToken . readTVar . updates
10183
102−createConfig :: P.Connection -> IO Config
103−createConfig conn = do
84+createConfig :: IO Config
85+createConfig = do
10486 mng <- newManager tlsManagerSettings
105− cfg <- atomically $ Config <$> return conn <*> return mng <*> newTVar M.empty <*> newTVar 0 <*> newTVar (CB.empty 20)
106− forkIO . forever $ threadDelay 7000000 >> (runHandler . createApp cfg $ startAddingPosts 25 cfg >>= uncurry sendUpdate)
87+ cfg <- atomically $ Config <$> newTVar 0 <*> newTVar 0 <*> newTVar M.empty <*> newTVar M.empty <*> return mng <*> newTVar M.empty <*> newTVar 0 <*> newTVar (CB.empty 20)
88+ forkIO . forever $ threadDelay 2000000 >> (runHandler . createApp cfg $ startAddingPosts 20 cfg >>= uncurry sendUpdate)
10789 return cfg
10890
10991type Token = Int
@@ -134,32 +116,18 @@ instance Semigroup Update where
134116instance Monoid Update where
135117 mempty = Update S.empty S.empty
136118
137−addConnection :: HasConfig c => c -> Socket -> STM SocketId
119+addConnection :: HasConfig c => c -> Connection -> STM ConnectionId
138120addConnection c conn = do
139121 let cfg = getCfg c
140− let sId = socketId conn
141− -- lastID <- fmap (+1) . readTVar . lastConnection $ cfg
142− -- flip writeTVar lastID . lastConnection $ cfg
143− modifyTVar (connections cfg) $ M.insert sId conn
144− return sId
122+ lastID <- fmap (+1) . readTVar . lastConnection $ cfg
123+ flip writeTVar lastID . lastConnection $ cfg
124+ modifyTVar (connections cfg) $ M.insert lastID conn
125+ return lastID
145126
146−removeConnection :: HasConfig c => c -> Socket -> STM ()
147−removeConnection c conn = do
127+removeConnection :: HasConfig c => c -> ConnectionId -> STM ()
128+removeConnection c cId = do
148129 let cfg = getCfg c
149− let sId = socketId conn
150− modifyTVar (connections cfg) $ M.delete sId
151−
152−
153−-- subscribe :: MonadIO m => Config -> Socket -> m ()
154−-- subscribe cfg conn = do
155−-- cId <- liftSTM $ addConnection cfg conn
156−-- -- liftIO $ flip finally (atomically $ removeConnection cfg cId) $ forever $ do
157−-- -- _ <- receiveDataMessage conn
158−-- -- return ()
159−-- return ()
160−
161−-- unsubscribe :: MonadIO m => Config -> Socket -> m ()
162−-- unsubscribe cfg conn = liftIO . atomically $ removeConnection cfg cId
130+ modifyTVar (connections cfg) $ M.delete cId
163131
164132
165133
@@ -171,9 +139,12 @@ type PostId = Int
171139type ConnectionId = Int
172140
173141data Config = Config {
174− dbConn :: P.Connection,
142+ lastUserID :: TVar UserId,
143+ lastPostID :: TVar PostId,
144+ users :: TVar (M.Map Int User),
145+ posts :: TVar (M.Map Int Post),
175146 manager :: Manager,
176− connections :: TVar (M.Map SocketId Socket),
147+ connections :: TVar (M.Map ConnectionId Connection),
177148 lastConnection :: TVar ConnectionId,
178149 updates :: TVar (CB.CyclicBuffer Update)
179150}
@@ -181,149 +152,94 @@ data Config = Config {
181152instance HasConfig Config where
182153 getCfg = id
183154
155+-- startAddingUsers :: Config -> App ()
156+-- startAddingUsers cfg = do
157+-- a <- liftIO $ loremBacon (manager cfg) (Just AllMeat) (Just 2) (Just False)
158+-- let t = fromMaybe ["None", "None"] a
159+-- liftIO . atomically $ newUser cfg (t !! 0) (t !! 1)
160+-- return ()
161+
184162startAddingPosts :: Int -> Config -> App (Token, Update)
185163startAddingPosts lim cfg = do
186164 cfg <- ask
187− let conn = dbConn cfg
188− -- let us = users socialDb
189− -- let ps = posts socialDb
165+ let us = users cfg
166+ let ps = posts cfg
190167
191168 a <- liftIO $ loremBacon (manager cfg) (Just AllMeat) (Just lim) (Just 0)
192169 g <- liftIO newStdGen
193170 let rns = randoms g
194−
195− let t = fromMaybe (repeat "None") a
196−
197− let deleteCount = 3
198171
199− (deletedIds, addedIds) <- join . liftIO . withSavepoint conn . return $ do
200− userIds <- liftIO . P.runBeamPostgres conn . runSelectReturningList . select $ do
201− uc <- all_ . users $ socialDb
202− pure (userId uc)
203−
204− if length userIds <= 0 then return ([],[])
205− else do
206− postIds <- liftIO . P.runBeamPostgres conn . runSelectReturningList . select $ do
207− uc <- all_ . posts $ socialDb
208− pure (postId uc)
209−
210−
211− let postCount = length postIds
212− let toDeleteIndecies = if postCount == 0 then [] else take deleteCount $ shuffle' postIds postCount g
213−
214− let toAdd = max 0 $ lim - length postIds + length toDeleteIndecies
215−
216− -- Just deletedPost <- liftIO . P.runBeamPostgres conn . runSelectReturningOne . select . limit_ 1 . offset_ (fromIntegral $ toDeleteIndex - 1) . all_ . posts $ socialDb
217− -- let deletedId = postId deletedPost
218−
219− unless (length toDeleteIndecies <= 0) $
220− liftIO . P.runBeamPostgres conn . runDelete $
221− delete (posts socialDb) ((`in_` map val_ toDeleteIndecies) . postId)
222−
223− nIds <- liftIO . P.runBeamPostgres conn . runInsertReturningList (posts socialDb) $ insertExpressions $ take toAdd . flip map (zip rns t) $ \(r, t) -> Post default_ (val_ (userIds %! r)) (val_ t)
224−
225− return (toDeleteIndecies, map postId nIds)
226−
227− liftIO . atomically $ do
228− let upd = Update (S.fromList addedIds) (S.fromList deletedIds)
229− let uppT = updates cfg
230− upp <- readTVar uppT
231− let (token, upp') = CB.insert upp upd
232− writeTVar uppT upp'
233− return (token, upd)
172+ let t = fromMaybe (repeat "None") a
173+ liftIO . atomically $ do
174+ uu <- readTVar us
175+ let uSize = M.size uu
176+ if uSize >= 1 then do
177+ pp <- readTVar ps
178+ let toDeleteIndex = head rns `mod` M.size pp
179+ let deletedKey = fst $ M.elemAt toDeleteIndex pp
180+ dd <- if M.size pp > 0 then writeTVar ps (M.deleteAt toDeleteIndex pp) >> return [deletedKey]
181+ else return []
182+ let toAdd = max 0 $ lim - (M.size pp - 1)
183+ add <- fmap (map postId) $ forM (zip rns $ take toAdd t) $ \(rnd, t) -> newPost cfg (randomFrom rnd uu) t
184+ let upd = Update (S.fromList add) (S.fromList dd)
185+ let uppT = updates cfg
186+ upp <- readTVar uppT
187+ let (token, upp') = CB.insert upp upd
188+ writeTVar uppT upp'
189+ return (token, upd)
190+ else return (0, mempty)
234191
235192sendUpdate :: Token -> Update -> App ()
236193sendUpdate t u = do
237194 cfg <- ask
238195 conns <- fmap M.elems . liftIO . readTVarIO $ connections cfg
239− forM conns $ \ conn -> emitTo conn postUpdates $ TokenizedResponse t u
240− -- liftIO $ forM conns $ flip sendTextData $ encode $ TokenizedResponse t u
196+ liftIO $ forM conns $ flip sendTextData $ encode $ TokenizedResponse t u
241197 return ()
242198
243−modIndex :: [a] -> Int -> a
244−modIndex aa i = aa !! (i `mod` length aa)
245−
246−infixl 5 %!
247−(%!) = modIndex
248−
249−
250199randomFrom :: Int -> M.Map k v -> k
251200randomFrom g as = M.keys as !! (abs g `mod` M.size as)
252201
253−newUser :: Config -> Text -> Text -> IO [User]
202+newUser :: Config -> Text -> Text -> STM User
254203newUser cfg first second = do
255− let conn = dbConn cfg
256− P.runBeamPostgres conn . runInsertReturningList (users socialDb) $ insertExpressions [User default_ (val_ first) (val_ second)]
257−
258−
259−-- newPost :: Config -> UserId -> Text -> STM Post
260−-- newPost cfg user body = do
261−-- let lastId = lastPostID cfg
262−-- let ps = posts cfg
263−-- do
264−-- id <- readTVar lastId
265−-- writeTVar lastId (id+1)
266−-- let p = Post id user body
267−-- modifyTVar ps $ M.insert id p
268−-- return p
269−
270−data UserT f = User {
271− userId :: Columnar f UserId,
272− firstName :: Columnar f Text,
273− secondName :: Columnar f Text
204+ let lastID = lastUserID cfg
205+ let us = users cfg
206+ do
207+ id <- readTVar lastID
208+ writeTVar lastID (id+1)
209+ let user = User (UserName first second) id
210+ oldUsers <- readTVar us
211+ writeTVar us $ M.insert id user oldUsers
212+ return user
213+
214+newPost :: Config -> UserId -> Text -> STM Post
215+newPost cfg user body = do
216+ let lastId = lastPostID cfg
217+ let ps = posts cfg
218+ do
219+ id <- readTVar lastId
220+ writeTVar lastId (id+1)
221+ let p = Post id user body
222+ modifyTVar ps $ M.insert id p
223+ return p
224+
225+data User = User {
226+ userName :: UserName,
227+ userId :: UserId
274228} deriving Generic
275229
276−instance Beamable UserT
277−
278−type User = UserT Identity
279−
280−instance Table UserT where
281− data PrimaryKey UserT f = UserId (Columnar f UserId)
282− deriving Generic
283−
284− primaryKey = UserId . userId
285−
286−instance Beamable (PrimaryKey UserT)
287−
288−data PostT f = Post {
289− postId :: Columnar f PostId,
290− postAuthor :: Columnar f UserId,
291− postBody :: Columnar f Text
230+data Post = Post {
231+ postId :: PostId,
232+ postAuthor :: UserId,
233+ postBody :: Text
292234} deriving Generic
293235
294−instance Beamable PostT
295−
296−type Post = PostT Identity
297−
298236instance ToJSON User
299237instance ToJSON Post
300238
301−instance Table PostT where
302− data PrimaryKey PostT f = PostId (Columnar f PostId)
303− deriving Generic
304−
305− primaryKey = PostId . postId
239+data UserName = UserName {
240+ firstName :: Text,
241+ secondName :: Text
242+} deriving (Generic)
306243
307−instance Beamable (PrimaryKey PostT)
308−
309−data SocialDb f = SocialDb {
310− users :: f (TableEntity UserT),
311− posts :: f (TableEntity PostT)
312−} deriving Generic
244+instance ToJSON UserName
313245
314−instance Database be SocialDb
315−
316−socialDb :: DatabaseSettings be SocialDb
317−socialDb = defaultDbSettings `withDbModification`
318− dbModification {
319− users = modifyTable id $ tableModification {
320− firstName = "firstname",
321− secondName = "secondname"
322− }
323− }
324−
325−subscribeWithToken :: Text
326−subscribeWithToken = "subscribeWithToken"
327−postUpdates :: Text
328−postUpdates = "postUpdates"
329−