@@ -1,5 +1,12 @@
11{-# LANGUAGE DeriveGeneric #-}
2+{-# LANGUAGE GADTs #-}
23{-# LANGUAGE OverloadedStrings #-}
4+{-# LANGUAGE FlexibleContexts #-}
5+{-# LANGUAGE FlexibleInstances #-}
6+{-# LANGUAGE TypeFamilies #-}
7+{-# LANGUAGE TypeApplications #-}
8+{-# LANGUAGE MultiParamTypeClasses #-}
9+
310
411module Foo.Data
512 (
@@ -9,7 +16,6 @@ module Foo.Data
916 createApp,
1017 createConfig,
1118 User(..),
12− UserName(..),
1319 Config,
1420 updates,
1521 allUsers,
@@ -17,12 +23,16 @@ module Foo.Data
1723 UserId,
1824 PostId,
1925 Post(..),
26+ postId,
2027 allPosts,
2128 addConnection,
2229 removeConnection,
2330 TokenizedResponse(..),
2431 Update(..),
25− getToken
32+ getToken,
33+ Token,
34+ subscribeWithToken,
35+ postUpdates
2636 ) where
2737
2838import GHC.Conc
@@ -50,6 +60,14 @@ import qualified Data.CyclicBuffer as CB
5060import Data.Set ((\\), union)
5161import Data.Monoid
5262import Debug.Trace
63+import Database.Beam
64+import qualified Database.Beam.Postgres as P
65+import Database.Beam.Query
66+import Database.PostgreSQL.Simple (withTransaction)
67+import Database.Beam.Backend.SQL.BeamExtensions
68+import System.Random.Shuffle
69+import Network.SocketIO
70+import Network.EngineIO (SocketId)
5371
5472type App = ReaderT Config Handler
5573
@@ -57,35 +75,35 @@ class HasConfig c where
5775 getCfg :: c -> Config
5876
5977allUsers :: App [User]
60−allUsers = ask >>= liftIO . fmap M.elems . readTVarIO . users
78+allUsers = asks dbConn >>= \conn -> liftIO . P.runBeamPostgres conn . runSelectReturningList . select . all_ . users $ socialDb
6179
6280post :: PostId -> App Post
6381post pId = do
64− cfg <- ask
65− u <- liftIO . atomically . fmap (M.lookup pId) . readTVar . posts $ cfg
82+ conn <- asks dbConn
83+ u <- liftIO . P.runBeamPostgres conn . runSelectReturningOne . flip lookup_ (PostId pId) . posts $ socialDb
6684 case u of
6785 Just u -> return u
6886 Nothing -> throwError err404
6987
7088user :: UserId -> App User
7189user uId = do
72− cfg <- ask
73− u <- liftIO . atomically . fmap (M.lookup uId) . readTVar . users $ cfg
90+ conn <- asks dbConn
91+ u <- liftIO . P.runBeamPostgres conn . runSelectReturningOne . flip lookup_ (UserId uId) . users $ socialDb
7492 case u of
7593 Just u -> return u
7694 Nothing -> throwError err404
7795
78−allPosts :: App (STM [Post])
79−allPosts = asks $ fmap M.elems . readTVar . posts
96+allPosts :: App [Post]
97+allPosts = asks dbConn >>= \conn -> liftIO . P.runBeamPostgres conn . runSelectReturningList . select . all_ . posts $ socialDb
8098
8199getToken :: App (STM Token)
82100getToken = asks $ fmap CB.lastToken . readTVar . updates
83101
84−createConfig :: IO Config
85−createConfig = do
102+createConfig :: P.Connection -> IO Config
103+createConfig conn = do
86104 mng <- newManager tlsManagerSettings
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)
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)
89107 return cfg
90108
91109type Token = Int
@@ -116,18 +134,32 @@ instance Semigroup Update where
116134instance Monoid Update where
117135 mempty = Update S.empty S.empty
118136
119−addConnection :: HasConfig c => c -> Connection -> STM ConnectionId
137+addConnection :: HasConfig c => c -> Socket -> STM SocketId
120138addConnection c conn = do
121139 let cfg = getCfg c
122− lastID <- fmap (+1) . readTVar . lastConnection $ cfg
123− flip writeTVar lastID . lastConnection $ cfg
124− modifyTVar (connections cfg) $ M.insert lastID conn
125− return lastID
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
126145
127−removeConnection :: HasConfig c => c -> ConnectionId -> STM ()
128−removeConnection c cId = do
146+removeConnection :: HasConfig c => c -> Socket -> STM ()
147+removeConnection c conn = do
129148 let cfg = getCfg c
130− modifyTVar (connections cfg) $ M.delete cId
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
131163
132164
133165
@@ -139,12 +171,9 @@ type PostId = Int
139171type ConnectionId = Int
140172
141173data Config = Config {
142− lastUserID :: TVar UserId,
143− lastPostID :: TVar PostId,
144− users :: TVar (M.Map Int User),
145− posts :: TVar (M.Map Int Post),
174+ dbConn :: P.Connection,
146175 manager :: Manager,
147− connections :: TVar (M.Map ConnectionId Connection),
176+ connections :: TVar (M.Map SocketId Socket),
148177 lastConnection :: TVar ConnectionId,
149178 updates :: TVar (CB.CyclicBuffer Update)
150179}
@@ -152,94 +181,149 @@ data Config = Config {
152181instance HasConfig Config where
153182 getCfg = id
154183
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−
162184startAddingPosts :: Int -> Config -> App (Token, Update)
163185startAddingPosts lim cfg = do
164186 cfg <- ask
165− let us = users cfg
166− let ps = posts cfg
187+ let conn = dbConn cfg
188+ -- let us = users socialDb
189+ -- let ps = posts socialDb
167190
168191 a <- liftIO $ loremBacon (manager cfg) (Just AllMeat) (Just lim) (Just 0)
169192 g <- liftIO newStdGen
170193 let rns = randoms g
171−
194+
172195 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)
196+
197+ let deleteCount = 3
198+
199+ (deletedIds, addedIds) <- join . liftIO . withTransaction 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)
191234
192235sendUpdate :: Token -> Update -> App ()
193236sendUpdate t u = do
194237 cfg <- ask
195238 conns <- fmap M.elems . liftIO . readTVarIO $ connections cfg
196− liftIO $ forM conns $ flip sendTextData $ encode $ TokenizedResponse t u
239+ forM conns $ \ conn -> emitTo conn postUpdates $ TokenizedResponse t u
240+ -- liftIO $ forM conns $ flip sendTextData $ encode $ TokenizedResponse t u
197241 return ()
198242
243+modIndex :: [a] -> Int -> a
244+modIndex aa i = aa !! (i `mod` length aa)
245+
246+infixl 5 %!
247+(%!) = modIndex
248+
249+
199250randomFrom :: Int -> M.Map k v -> k
200251randomFrom g as = M.keys as !! (abs g `mod` M.size as)
201252
202−newUser :: Config -> Text -> Text -> STM User
253+newUser :: Config -> Text -> Text -> IO [User]
203254newUser cfg first second = do
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
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
228274} deriving Generic
229275
230−data Post = Post {
231− postId :: PostId,
232− postAuthor :: UserId,
233− postBody :: Text
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
234292} deriving Generic
235293
294+instance Beamable PostT
295+
296+type Post = PostT Identity
297+
236298instance ToJSON User
237299instance ToJSON Post
238300
239−data UserName = UserName {
240− firstName :: Text,
241− secondName :: Text
242−} deriving (Generic)
301+instance Table PostT where
302+ data PrimaryKey PostT f = PostId (Columnar f PostId)
303+ deriving Generic
304+
305+ primaryKey = PostId . postId
243306
244−instance ToJSON UserName
307+instance Beamable (PrimaryKey PostT)
308+
309+data SocialDb f = SocialDb {
310+ users :: f (TableEntity UserT),
311+ posts :: f (TableEntity PostT)
312+} deriving Generic
245313
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+