aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--lib/Server.hs349
-rw-r--r--lib/Server/Base.hs9
-rw-r--r--lib/Server/Ingest.hs211
-rw-r--r--lib/Server/Subscribe.hs63
-rw-r--r--lib/Server/Util.hs64
5 files changed, 394 insertions, 302 deletions
diff --git a/lib/Server.hs b/lib/Server.hs
index 73c55cb..1833aa0 100644
--- a/lib/Server.hs
+++ b/lib/Server.hs
@@ -1,97 +1,54 @@
-{-# LANGUAGE DataKinds #-}
-{-# LANGUAGE ExplicitNamespaces #-}
-{-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE PartialTypeSignatures #-}
{-# LANGUAGE RecordWildCards #-}
-- Implementation of the API. This module is the main point of the program.
module Server (application) where
-import Control.Concurrent.STM (TQueue, TVar, atomically,
- newTQueue, newTVar,
- newTVarIO, readTQueue,
- readTVar, writeTQueue,
- writeTVar)
-import Control.Monad (forever, unless, void,
- when)
-import Control.Monad.Catch (handle)
-import Control.Monad.Extra (ifM, mapMaybeM, maybeM,
- unlessM, whenJust, whenM)
-import Control.Monad.IO.Class (MonadIO (liftIO))
-import Control.Monad.Logger (LoggingT, MonadLogger,
- NoLoggingT, logInfoN,
- logWarnN)
-import Control.Monad.Reader (MonadReader, ReaderT,
- forM)
-import Control.Monad.Trans (lift)
-import Data.Aeson ((.=))
-import qualified Data.Aeson as A
-import qualified Data.ByteString.Char8 as C8
-import Data.Coerce (coerce)
-import Data.Functor ((<&>))
-import qualified Data.Map as M
-import Data.Pool (Pool)
-import Data.Proxy (Proxy (Proxy))
-import Data.Swagger (toSchema)
-import Data.Text (Text)
-import Data.Text.Encoding (decodeASCII, decodeUtf8)
-import Data.Time (NominalDiffTime,
- UTCTime (utctDay),
- addUTCTime, diffUTCTime,
- getCurrentTime,
- nominalDay)
-import qualified Data.Vector as V
-import Database.Persist
-import Database.Persist.Postgresql (SqlBackend,
- migrateEnableExtension,
- runMigration)
-import Fmt ((+|), (|+))
-import qualified Network.WebSockets as WS
-import Servant (Application,
- ServerError (errBody),
- err400, err401, err404,
- serve,
- serveDirectoryFileServer,
- throwError)
-import Servant.API (NoContent (..),
- (:<|>) (..))
-import Servant.Server (Handler, hoistServer)
-import Servant.Swagger (toSwagger)
-
-import API
+import API (API, CompleteAPI, Metrics (..))
+import Conduit (ResourceT)
+import Config (LoggingConfig, ServerConfig (..))
+import Control.Concurrent.STM (newTVarIO)
+import Control.Monad.Extra (forM)
+import Control.Monad.IO.Class (MonadIO (liftIO))
+import Control.Monad.Logger (MonadLogger)
+import Control.Monad.Reader (ReaderT)
+import Data.ByteString.Lazy (toStrict)
+import Data.Functor ((<&>))
+import qualified Data.Map as M
+import Data.Pool (Pool)
+import Data.Proxy (Proxy (Proxy))
+import Data.Text.Encoding (decodeUtf8)
+import Data.Time (getCurrentTime)
+import Data.UUID (UUID)
+import Database.Persist (Entity (..),
+ PersistQueryRead (selectFirst),
+ SelectOpt (Desc), selectList,
+ (<-.), (==.), (>=.), (||.))
+import Database.Persist.Postgresql (SqlBackend,
+ migrateEnableExtension,
+ runMigration)
+import Fmt ((+|), (|+))
import qualified GTFS
import Persist
-import Server.ControlRoom
-import Server.GTFS_RT (gtfsRealtimeServer)
-import Server.Util (Service, ServiceM,
- runService, sendErrorMsg,
- utcToSeconds)
-import Yesod (toWaiAppPlain)
+import Prometheus (Info (Info), exportMetricsAsText,
+ gauge, register)
+import Prometheus.Metric.GHC (ghcMetrics)
+import Servant (Application, err401, serve,
+ serveDirectoryFileServer,
+ throwError)
+import Servant.API ((:<|>) (..))
+import Servant.Server (hoistServer)
+import Servant.Swagger (toSwagger)
+import Server.Base (ServerState)
+import Server.ControlRoom (ControlRoom (ControlRoom))
+import Server.GTFS_RT (gtfsRealtimeServer)
+import Server.Ingest (handleTrackerRegister,
+ handleTrainPing, handleWS)
+import Server.Subscribe (handleSubscribe)
+import Server.Util (Service, runService)
+import System.IO.Unsafe (unsafePerformIO)
+import Yesod (toWaiAppPlain)
-import Conduit (ResourceT)
-import Config (LoggingConfig,
- ServerConfig (..))
-import Control.Exception (throw)
-import Data.ByteString (ByteString)
-import Data.ByteString.Lazy (toStrict)
-import Data.Foldable (minimumBy)
-import Data.Function (on, (&))
-import Data.Maybe (fromMaybe)
-import qualified Data.Text as T
-import Data.Time.LocalTime.TimeZone.Olson (getTimeZoneSeriesFromOlsonFile)
-import Data.Time.LocalTime.TimeZone.Series (TimeZoneSeries)
-import Data.UUID (UUID)
-import qualified Data.UUID as UUID
-import Extrapolation (Extrapolator (..),
- LinearExtrapolator (..),
- euclid)
-import GTFS (Seconds (unSeconds),
- seconds2Double)
-import Prometheus
-import Prometheus.Metric.GHC
-import System.FilePath ((</>))
-import System.IO.Unsafe
application :: GTFS.GTFS -> Pool SqlBackend -> ServerConfig -> IO Application
application gtfs dbpool settings = do
@@ -100,189 +57,48 @@ application gtfs dbpool settings = do
<$> register (gauge (Info "ws_connections" "Number of WS Connections"))
register ghcMetrics
- -- TODO: maybe cache these in a TVar, we're not likely to ever need
- -- more than one of these
- let getTzseries tzname = getTimeZoneSeriesFromOlsonFile
- (serverConfigZoneinfoPath settings </> T.unpack tzname)
-
subscribers <- newTVarIO mempty
pure $ serve (Proxy @CompleteAPI)
$ hoistServer (Proxy @CompleteAPI) (runService (serverConfigLogging settings))
- $ server gtfs getTzseries metrics subscribers dbpool settings
+ $ server gtfs metrics subscribers dbpool settings
--- databaseMigration :: ConnectionString -> IO ()
doMigration pool = runSqlWithoutLog pool $ runMigration $ do
migrateEnableExtension "uuid-ossp"
migrateAll
-server :: GTFS.GTFS -> (Text -> IO TimeZoneSeries) -> Metrics -> TVar (M.Map UUID [TQueue (Maybe TrainPing)]) -> Pool SqlBackend -> ServerConfig -> Service CompleteAPI
-server gtfs getTzseries Metrics{..} subscribers dbpool settings = handleDebugAPI
- :<|> (handleTrackerRegister :<|> handleTrainPing (throwError err401) :<|> handleWS
- :<|> handleSubscribe :<|> handleDebugState :<|> handleDebugTrain
+server
+ :: GTFS.GTFS
+ -> Metrics
+ -> ServerState
+ -> Pool SqlBackend
+ -> ServerConfig
+ -> Service CompleteAPI
+server gtfs metrics@Metrics{..} subscribers dbpool settings = handleDebugAPI
+ :<|> (handleTrackerRegister dbpool
+ :<|> handleTrainPing dbpool subscribers settings (throwError err401)
+ :<|> handleWS dbpool subscribers settings metrics
+ :<|> handleSubscribe dbpool subscribers
+ :<|> handleDebugState :<|> handleDebugTrain
:<|> pure (GTFS.gtfsFile gtfs) :<|> gtfsRealtimeServer gtfs dbpool)
- :<|> metrics
+ :<|> handleMetrics
:<|> serveDirectoryFileServer (serverConfigAssets settings)
:<|> pure (unsafePerformIO (toWaiAppPlain (ControlRoom gtfs dbpool settings)))
- where handleTrackerRegister RegisterJson{..} = do
- today <- liftIO getCurrentTime <&> utctDay
- expires <- liftIO $ getCurrentTime <&> addUTCTime validityPeriod
- runSql dbpool $ do
- TrackerKey tracker <- insert (Tracker expires False registerAgent Nothing)
- pure tracker
- handleTrainPing onError ping@TrainPing{..} =
- isTokenValid dbpool trainPingToken >>= \case
- Nothing -> onError >> pure Nothing
- Just tracker@Tracker{..} -> do
-
- -- if the tracker is not associated with a ticket, it is probably
- -- just starting out on a new trip, or has finished an old one.
- maybeTicketId <- case trackerCurrentTicket of
- Just ticketId -> pure (Just ticketId)
- Nothing -> runSql dbpool $ do
- now <- liftIO getCurrentTime
- tickets <- selectList [ TicketDay ==. utctDay now, TicketCompleted ==. False ] []
- ticketsWithFirstStation <- flip mapMaybeM tickets
- (\ticket@(Entity ticketId _) -> do
- selectFirst [StopTicket ==. ticketId] [Asc StopSequence] >>= \case
- Nothing -> pure Nothing
- Just (Entity _ stop) -> do
- station <- getJust (stopStation stop)
- tzseries <- liftIO $ getTzseries (GTFS.tzname (stopDeparture stop))
- pure (Just (ticket, station, stop, tzseries)))
-
- if null ticketsWithFirstStation then pure Nothing else do
- let (closestTicket, _, _, _) = minimumBy
- -- (compare `on` euclid trainPingGeopos . stationGeopos . snd)
- (compare `on`
- (\(Entity _ ticket, station, stop, tzseries) ->
- let
- runningDay = ticketDay ticket
- spaceDistance = euclid trainPingGeopos (stationGeopos station)
- timeDiff =
- GTFS.toSeconds (stopDeparture stop) tzseries runningDay
- - utcToSeconds now runningDay
- in
- euclid trainPingGeopos (stationGeopos station)
- + abs (seconds2Double timeDiff / 3600)))
- ticketsWithFirstStation
- logInfoN
- $ "Tracker "+|UUID.toString (coerce trainPingToken)|+
- " is now handling ticket "+|UUID.toString (coerce (entityKey closestTicket))|+
- " (trip "+|ticketTripName (entityVal closestTicket)|+")."
-
- update (coerce trainPingToken)
- [TrackerCurrentTicket =. Just (entityKey closestTicket)]
-
- pure (Just (entityKey closestTicket))
-
- ticketId <- case maybeTicketId of
- Just ticketId -> pure ticketId
- Nothing -> do
- logWarnN $ "Tracker "+|UUID.toString (coerce trainPingToken)|+
- " sent a ping, but no trips are running today."
- throwError err400
-
- runSql dbpool $ do
- ticket@Ticket{..} <- getJust ticketId
-
- stations <- selectList [ StopTicket ==. ticketId ] [Asc StopArrival]
- >>= mapM (\stop -> do
- station <- getJust (stopStation (entityVal stop))
- tzseries <- liftIO $ getTzseries (GTFS.tzname (stopArrival (entityVal stop)))
- pure (entityVal stop, station, tzseries))
- <&> V.fromList
-
- shapePoints <- selectList [ShapePointShape ==. ticketShape] [Asc ShapePointIndex]
- <&> (V.fromList . fmap entityVal)
-
- let anchor = extrapolateAnchorFromPing LinearExtrapolator
- ticketId ticket stations shapePoints ping
-
- insert ping
-
- last <- selectFirst [TrainAnchorTicket ==. ticketId] [Desc TrainAnchorWhen]
- -- only insert new estimates if they've actually changed anything
- when (fmap (trainAnchorDelay . entityVal) last /= Just (trainAnchorDelay anchor))
- $ void $ insert anchor
-
- -- are we at the final stop? if so, mark this ticket as done
- -- & the tracker as free
- let maxSequence = V.last stations
- & (\(stop, _, _) -> stopSequence stop)
- & fromIntegral
- when (trainAnchorSequence anchor + 0.1 >= maxSequence) $ do
- update (coerce trainPingToken)
- [TrackerCurrentTicket =. Nothing]
- update ticketId
- [TicketCompleted =. True]
- logInfoN $ "Tracker "+|UUID.toString (coerce trainPingToken)|+
- " has completed ticket "+|UUID.toString (coerce ticketId)|+
- " (trip "+|ticketTripName|+")"
-
- queues <- liftIO $ atomically $ do
- queues <- readTVar subscribers <&> M.lookup (coerce ticketId)
- whenJust queues $
- mapM_ (\q -> writeTQueue q (Just ping))
- pure queues
- pure (Just anchor)
- handleWS conn = do
- liftIO $ WS.forkPingThread conn 30
- incGauge metricsWSGauge
- handle (\(e :: WS.ConnectionException) -> decGauge metricsWSGauge) $ forever $ do
- msg <- liftIO $ WS.receiveData conn
- case A.eitherDecode msg of
- Left err -> do
- logWarnN ("stray websocket message: "+|decodeASCII (toStrict msg)|+" (could not decode: "+|err|+")")
- liftIO $ WS.sendClose conn (C8.pack err)
- -- TODO: send a close msg (Nothing) to the subscribed queues? decGauge metricsWSGauge
- Right ping -> do
- -- if invalid token, send a "polite" close request. Note that the client may
- -- ignore this and continue sending messages, which will continue to be handled.
- handleTrainPing (liftIO $ WS.sendClose conn ("" :: ByteString)) ping >>= \case
- Just anchor -> liftIO $ WS.sendTextData conn (A.encode anchor)
- Nothing -> pure ()
- handleSubscribe (ticketId :: UUID) conn = liftIO $ WS.withPingThread conn 30 (pure ()) $ do
- queue <- atomically $ do
- queue <- newTQueue
- qs <- readTVar subscribers
- writeTVar subscribers
- $ M.insertWith (<>) ticketId [queue] qs
- pure queue
- -- send most recent ping, if any (so we won't have to wait for movement)
- lastPing <- runSqlWithoutLog dbpool $ do
- trackers <- getTicketTrackers ticketId
- <&> fmap entityKey
- selectFirst [TrainPingToken <-. trackers] [Desc TrainPingTimestamp]
- <&> fmap entityVal
- whenJust lastPing $ \ping ->
- WS.sendTextData conn (A.encode lastPing)
- handle (\(e :: WS.ConnectionException) -> removeSubscriber queue) $ forever $ do
- res <- atomically $ readTQueue queue
- case res of
- Just ping -> WS.sendTextData conn (A.encode ping)
- Nothing -> do
- removeSubscriber queue
- WS.sendClose conn (C8.pack "train ended")
- where removeSubscriber queue = atomically $ do
- qs <- readTVar subscribers
- writeTVar subscribers
- $ M.adjust (filter (/= queue)) ticketId qs
- handleDebugState = do
- now <- liftIO getCurrentTime
- runSql dbpool $ do
- tracker <- selectList [TrackerBlocked ==. False, TrackerExpires >=. now] []
- pairs <- forM tracker $ \(Entity token@(TrackerKey uuid) _) -> do
- entities <- selectList [TrainPingToken ==. token] []
- pure (uuid, fmap entityVal entities)
- pure (M.fromList pairs)
- handleDebugTrain ticketId = do
- runSql dbpool $ do
- trackers <- getTicketTrackers ticketId
- pings <- forM trackers $ \(Entity token _) -> do
- selectList [TrainPingToken ==. token] [] <&> fmap entityVal
- pure (concat pings)
- handleDebugAPI = pure $ toSwagger (Proxy @API)
- metrics = exportMetricsAsText <&> (decodeUtf8 . toStrict)
+ where
+ handleDebugState = do
+ now <- liftIO getCurrentTime
+ runSql dbpool $ do
+ tracker <- selectList [TrackerBlocked ==. False, TrackerExpires >=. now] []
+ pairs <- forM tracker $ \(Entity token@(TrackerKey uuid) _) -> do
+ entities <- selectList [TrainPingToken ==. token] []
+ pure (uuid, fmap entityVal entities)
+ pure (M.fromList pairs)
+ handleDebugTrain ticketId = runSql dbpool $ do
+ trackers <- getTicketTrackers ticketId
+ pings <- forM trackers $ \(Entity token _) -> do
+ selectList [TrainPingToken ==. token] [] <&> fmap entityVal
+ pure (concat pings)
+ handleDebugAPI = pure $ toSwagger (Proxy @API)
+ handleMetrics = exportMetricsAsText <&> (decodeUtf8 . toStrict)
getTicketTrackers :: (MonadLogger (t (ResourceT IO)), MonadIO (t (ResourceT IO)))
=> UUID -> ReaderT SqlBackend (t (ResourceT IO)) [Entity Tracker]
@@ -290,22 +106,3 @@ getTicketTrackers ticketId = do
joins <- selectList [TrackerTicketTicket ==. TicketKey ticketId] []
<&> fmap (trackerTicketTracker . entityVal)
selectList ([TrackerId <-. joins] ||. [TrackerCurrentTicket ==. Just (TicketKey ticketId)]) []
-
-
--- TODO: proper debug logging for expired tokens
-isTokenValid :: Pool SqlBackend -> TrackerId -> ServiceM (Maybe Tracker)
-isTokenValid dbpool token = runSql dbpool $ get token >>= \case
- Just tracker | not (trackerBlocked tracker) -> do
- ifM (hasExpired (trackerExpires tracker))
- (pure Nothing)
- (pure (Just tracker))
- _ -> pure Nothing
-
-hasExpired :: MonadIO m => UTCTime -> m Bool
-hasExpired limit = do
- now <- liftIO getCurrentTime
- pure (now > limit)
-
-validityPeriod :: NominalDiffTime
-validityPeriod = nominalDay
-
diff --git a/lib/Server/Base.hs b/lib/Server/Base.hs
new file mode 100644
index 0000000..14b77ca
--- /dev/null
+++ b/lib/Server/Base.hs
@@ -0,0 +1,9 @@
+
+module Server.Base (ServerState) where
+
+import Control.Concurrent.STM (TQueue, TVar)
+import qualified Data.Map as M
+import Data.UUID (UUID)
+import Persist (TrainPing)
+
+type ServerState = TVar (M.Map UUID [TQueue (Maybe TrainPing)])
diff --git a/lib/Server/Ingest.hs b/lib/Server/Ingest.hs
new file mode 100644
index 0000000..d774017
--- /dev/null
+++ b/lib/Server/Ingest.hs
@@ -0,0 +1,211 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE RecordWildCards #-}
+
+module Server.Ingest (handleTrackerRegister, handleTrainPing, handleWS) where
+import API (Metrics (..),
+ RegisterJson (..))
+import Control.Concurrent.STM (atomically, readTVar,
+ writeTQueue)
+import Control.Monad (forever, void, when)
+import Control.Monad.Catch (handle)
+import Control.Monad.Extra (ifM, mapMaybeM, whenJust)
+import Control.Monad.IO.Class (MonadIO (liftIO))
+import Control.Monad.Logger (LoggingT, logInfoN,
+ logWarnN)
+import Control.Monad.Reader (ReaderT)
+import qualified Data.Aeson as A
+import qualified Data.ByteString.Char8 as C8
+import Data.Coerce (coerce)
+import Data.Functor ((<&>))
+import qualified Data.Map as M
+import Data.Pool (Pool)
+import Data.Text (Text)
+import Data.Text.Encoding (decodeASCII, decodeUtf8)
+import Data.Time (NominalDiffTime,
+ UTCTime (..), addUTCTime,
+ getCurrentTime,
+ nominalDay)
+import qualified Data.Vector as V
+import Database.Persist
+import Database.Persist.Postgresql (SqlBackend)
+import Fmt ((+|), (|+))
+import qualified GTFS
+import qualified Network.WebSockets as WS
+import Persist
+import Servant (err400, throwError)
+import Servant.Server (Handler)
+import Server.Util (ServiceM, getTzseries,
+ utcToSeconds)
+
+import Config (LoggingConfig,
+ ServerConfig)
+import Data.ByteString (ByteString)
+import Data.ByteString.Lazy (toStrict)
+import Data.Foldable (minimumBy)
+import Data.Function (on, (&))
+import qualified Data.Text as T
+import Data.Time.LocalTime.TimeZone.Series (TimeZoneSeries)
+import qualified Data.UUID as UUID
+import Extrapolation (Extrapolator (..),
+ LinearExtrapolator (..),
+ euclid)
+import GTFS (seconds2Double)
+import Prometheus (decGauge, incGauge)
+import Server.Base (ServerState)
+
+
+
+handleTrackerRegister
+ :: Pool SqlBackend
+ -> RegisterJson
+ -> ServiceM Token
+handleTrackerRegister dbpool RegisterJson{..} = do
+ today <- liftIO getCurrentTime <&> utctDay
+ expires <- liftIO $ getCurrentTime <&> addUTCTime validityPeriod
+ runSql dbpool $ do
+ TrackerKey tracker <- insert (Tracker expires False registerAgent Nothing)
+ pure tracker
+ where
+ validityPeriod :: NominalDiffTime
+ validityPeriod = nominalDay
+
+handleTrainPing
+ :: Pool SqlBackend
+ -> ServerState
+ -> ServerConfig
+ -> LoggingT (ReaderT LoggingConfig Handler) a
+ -> TrainPing
+ -> LoggingT (ReaderT LoggingConfig Handler) (Maybe TrainAnchor)
+handleTrainPing dbpool subscribers cfg onError ping@TrainPing{..} =
+ isTokenValid dbpool trainPingToken >>= \case
+ Nothing -> onError >> pure Nothing
+ Just tracker@Tracker{..} -> do
+
+ -- if the tracker is not associated with a ticket, it is probably
+ -- just starting out on a new trip, or has finished an old one.
+ maybeTicketId <- case trackerCurrentTicket of
+ Just ticketId -> pure (Just ticketId)
+ Nothing -> runSql dbpool $ do
+ now <- liftIO getCurrentTime
+ tickets <- selectList [ TicketDay ==. utctDay now, TicketCompleted ==. False ] []
+ ticketsWithFirstStation <- flip mapMaybeM tickets
+ (\ticket@(Entity ticketId _) -> do
+ selectFirst [StopTicket ==. ticketId] [Asc StopSequence] >>= \case
+ Nothing -> pure Nothing
+ Just (Entity _ stop) -> do
+ station <- getJust (stopStation stop)
+ tzseries <- liftIO $ getTzseries cfg (GTFS.tzname (stopDeparture stop))
+ pure (Just (ticket, station, stop, tzseries)))
+
+ if null ticketsWithFirstStation then pure Nothing else do
+ let (closestTicket, _, _, _) = minimumBy
+ -- (compare `on` euclid trainPingGeopos . stationGeopos . snd)
+ (compare `on`
+ (\(Entity _ ticket, station, stop, tzseries) ->
+ let
+ runningDay = ticketDay ticket
+ spaceDistance = euclid trainPingGeopos (stationGeopos station)
+ timeDiff =
+ GTFS.toSeconds (stopDeparture stop) tzseries runningDay
+ - utcToSeconds now runningDay
+ in
+ euclid trainPingGeopos (stationGeopos station)
+ + abs (seconds2Double timeDiff / 3600)))
+ ticketsWithFirstStation
+ logInfoN
+ $ "Tracker "+|UUID.toString (coerce trainPingToken)|+
+ " is now handling ticket "+|UUID.toString (coerce (entityKey closestTicket))|+
+ " (trip "+|ticketTripName (entityVal closestTicket)|+")."
+
+ update trainPingToken
+ [TrackerCurrentTicket =. Just (entityKey closestTicket)]
+
+ pure (Just (entityKey closestTicket))
+
+ ticketId <- case maybeTicketId of
+ Just ticketId -> pure ticketId
+ Nothing -> do
+ logWarnN $ "Tracker "+|UUID.toString (coerce trainPingToken)|+
+ " sent a ping, but no trips are running today."
+ throwError err400
+
+ runSql dbpool $ do
+ ticket@Ticket{..} <- getJust ticketId
+
+ stations <- selectList [ StopTicket ==. ticketId ] [Asc StopArrival]
+ >>= mapM (\stop -> do
+ station <- getJust (stopStation (entityVal stop))
+ tzseries <- liftIO $ getTzseries cfg (GTFS.tzname (stopArrival (entityVal stop)))
+ pure (entityVal stop, station, tzseries))
+ <&> V.fromList
+
+ shapePoints <- selectList [ShapePointShape ==. ticketShape] [Asc ShapePointIndex]
+ <&> (V.fromList . fmap entityVal)
+
+ let anchor = extrapolateAnchorFromPing LinearExtrapolator
+ ticketId ticket stations shapePoints ping
+
+ insert ping
+
+ last <- selectFirst [TrainAnchorTicket ==. ticketId] [Desc TrainAnchorWhen]
+ -- only insert new estimates if they've actually changed anything
+ when (fmap (trainAnchorDelay . entityVal) last /= Just (trainAnchorDelay anchor))
+ $ void $ insert anchor
+
+ -- are we at the final stop? if so, mark this ticket as done
+ -- & the tracker as free
+ let maxSequence = V.last stations
+ & (\(stop, _, _) -> stopSequence stop)
+ & fromIntegral
+ when (trainAnchorSequence anchor + 0.1 >= maxSequence) $ do
+ update trainPingToken
+ [TrackerCurrentTicket =. Nothing]
+ update ticketId
+ [TicketCompleted =. True]
+ logInfoN $ "Tracker "+|UUID.toString (coerce trainPingToken)|+
+ " has completed ticket "+|UUID.toString (coerce ticketId)|+
+ " (trip "+|ticketTripName|+")"
+
+ queues <- liftIO $ atomically $ do
+ queues <- readTVar subscribers <&> M.lookup (coerce ticketId)
+ whenJust queues $
+ mapM_ (\q -> writeTQueue q (Just ping))
+ pure queues
+ pure (Just anchor)
+
+handleWS
+ :: Pool SqlBackend
+ -> ServerState
+ -> ServerConfig
+ -> Metrics
+ -> WS.Connection -> ServiceM ()
+handleWS dbpool subscribers cfg Metrics{..} conn = do
+ liftIO $ WS.forkPingThread conn 30
+ incGauge metricsWSGauge
+ handle (\(e :: WS.ConnectionException) -> decGauge metricsWSGauge) $ forever $ do
+ msg <- liftIO $ WS.receiveData conn
+ case A.eitherDecode msg of
+ Left err -> do
+ logWarnN ("stray websocket message: "+|decodeASCII (toStrict msg)|+" (could not decode: "+|err|+")")
+ liftIO $ WS.sendClose conn (C8.pack err)
+ -- TODO: send a close msg (Nothing) to the subscribed queues? decGauge metricsWSGauge
+ Right ping -> do
+ -- if invalid token, send a "polite" close request. Note that the client may
+ -- ignore this and continue sending messages, which will continue to be handled.
+ handleTrainPing dbpool subscribers cfg (liftIO $ WS.sendClose conn ("" :: ByteString)) ping >>= \case
+ Just anchor -> liftIO $ WS.sendTextData conn (A.encode anchor)
+ Nothing -> pure ()
+
+-- TODO: proper debug logging for expired tokens
+isTokenValid :: Pool SqlBackend -> TrackerId -> ServiceM (Maybe Tracker)
+isTokenValid dbpool token = runSql dbpool $ get token >>= \case
+ Just tracker | not (trackerBlocked tracker) -> do
+ ifM (hasExpired (trackerExpires tracker))
+ (pure Nothing)
+ (pure (Just tracker))
+ _ -> pure Nothing
+
+hasExpired :: MonadIO m => UTCTime -> m Bool
+hasExpired limit = do
+ now <- liftIO getCurrentTime
+ pure (now > limit)
diff --git a/lib/Server/Subscribe.hs b/lib/Server/Subscribe.hs
new file mode 100644
index 0000000..fdc092b
--- /dev/null
+++ b/lib/Server/Subscribe.hs
@@ -0,0 +1,63 @@
+
+module Server.Subscribe where
+import Conduit (MonadIO (..))
+import Control.Concurrent.STM (atomically, newTQueue, readTQueue,
+ readTVar, writeTVar)
+import Control.Exception (handle)
+import Control.Monad.Extra (forever, whenJust)
+import qualified Data.Aeson as A
+import qualified Data.ByteString.Char8 as C8
+import Data.Functor ((<&>))
+import Data.Map (Map)
+import qualified Data.Map as M
+import Data.Pool
+import Data.UUID (UUID)
+import Database.Persist (Entity (entityKey), SelectOpt (Desc),
+ entityVal, selectFirst, selectList,
+ (<-.), (==.), (||.))
+import Database.Persist.Sql (SqlBackend)
+import qualified Network.WebSockets as WS
+import Persist
+import Server.Base (ServerState)
+import Server.Util (ServiceM)
+
+
+handleSubscribe
+ :: Pool SqlBackend
+ -> ServerState
+ -> UUID
+ -> WS.Connection
+ -> ServiceM ()
+handleSubscribe dbpool subscribers (ticketId :: UUID) conn = liftIO $ WS.withPingThread conn 30 (pure ()) $ do
+ queue <- atomically $ do
+ queue <- newTQueue
+ qs <- readTVar subscribers
+ writeTVar subscribers
+ $ M.insertWith (<>) ticketId [queue] qs
+ pure queue
+ -- send most recent ping, if any (so we won't have to wait for movement)
+ lastPing <- runSqlWithoutLog dbpool $ do
+ trackers <- getTicketTrackers ticketId
+ <&> fmap entityKey
+ selectFirst [TrainPingToken <-. trackers] [Desc TrainPingTimestamp]
+ <&> fmap entityVal
+ whenJust lastPing $ \ping ->
+ WS.sendTextData conn (A.encode lastPing)
+ handle (\(e :: WS.ConnectionException) -> removeSubscriber queue) $ forever $ do
+ res <- atomically $ readTQueue queue
+ case res of
+ Just ping -> WS.sendTextData conn (A.encode ping)
+ Nothing -> do
+ removeSubscriber queue
+ WS.sendClose conn (C8.pack "train ended")
+ where removeSubscriber queue = atomically $ do
+ qs <- readTVar subscribers
+ writeTVar subscribers
+ $ M.adjust (filter (/= queue)) ticketId qs
+
+-- getTicketTrackers :: (MonadLogger (t (ResourceT IO)), MonadIO (t (ResourceT IO)))
+-- => UUID -> ReaderT SqlBackend (t (ResourceT IO)) [Entity Tracker]
+getTicketTrackers ticketId = do
+ joins <- selectList [TrackerTicketTicket ==. TicketKey ticketId] []
+ <&> fmap (trackerTicketTracker . entityVal)
+ selectList ([TrackerId <-. joins] ||. [TrackerCurrentTicket ==. Just (TicketKey ticketId)]) []
diff --git a/lib/Server/Util.hs b/lib/Server/Util.hs
index 0106428..290b9c5 100644
--- a/lib/Server/Util.hs
+++ b/lib/Server/Util.hs
@@ -1,33 +1,41 @@
{-# LANGUAGE BlockArguments #-}
{-# LANGUAGE RecordWildCards #-}
-- | mostly the monad the service runs in
-module Server.Util (Service, ServiceM, runService, sendErrorMsg, secondsNow, utcToSeconds, runLogging) where
+module Server.Util (Service, ServiceM, runService, sendErrorMsg, secondsNow, utcToSeconds, runLogging, getTzseries) where
-import Config (LoggingConfig (..))
-import Control.Exception (handle, try)
-import Control.Monad.Extra (void, whenJust)
-import Control.Monad.IO.Class (MonadIO (liftIO))
-import Control.Monad.Logger (Loc, LogLevel (..), LogSource, LogStr,
- LoggingT (..), defaultOutput,
- fromLogStr, runStderrLoggingT)
-import Control.Monad.Reader (ReaderT (..))
-import qualified Data.Aeson as A
-import Data.ByteString (ByteString)
-import qualified Data.ByteString as C8
-import Data.Text (Text)
-import qualified Data.Text as T
-import Data.Text.Encoding (decodeUtf8Lenient)
-import Data.Time (Day, UTCTime (..), diffUTCTime,
- getCurrentTime,
- nominalDiffTimeToSeconds)
-import Fmt ((+|), (|+))
-import GHC.IO.Exception (IOException (IOError))
-import GTFS (Seconds (..))
-import Prometheus (MonadMonitor (doIO))
-import Servant (Handler, ServerError, ServerT, err404,
- errBody, errHeaders, throwError)
-import System.IO (stderr)
-import System.Process.Extra (callProcess)
+import Config (LoggingConfig (..),
+ ServerConfig (..))
+import Control.Exception (handle, try)
+import Control.Monad.Extra (void, whenJust)
+import Control.Monad.IO.Class (MonadIO (liftIO))
+import Control.Monad.Logger (Loc, LogLevel (..),
+ LogSource, LogStr,
+ LoggingT (..),
+ defaultOutput, fromLogStr,
+ runStderrLoggingT)
+import Control.Monad.Reader (ReaderT (..))
+import qualified Data.Aeson as A
+import Data.ByteString (ByteString)
+import qualified Data.ByteString as C8
+import Data.Text (Text)
+import qualified Data.Text as T
+import Data.Text.Encoding (decodeUtf8Lenient)
+import Data.Time (Day, UTCTime (..),
+ diffUTCTime,
+ getCurrentTime,
+ nominalDiffTimeToSeconds)
+import Data.Time.LocalTime.TimeZone.Olson (getTimeZoneSeriesFromOlsonFile)
+import Data.Time.LocalTime.TimeZone.Series (TimeZoneSeries)
+import Fmt ((+|), (|+))
+import GHC.IO.Exception (IOException (IOError))
+import GTFS (Seconds (..))
+import Prometheus (MonadMonitor (doIO))
+import Servant (Handler, ServerError,
+ ServerT, err404, errBody,
+ errHeaders, throwError)
+import System.FilePath ((</>))
+import System.IO (stderr)
+import System.Process.Extra (callProcess)
type ServiceM = LoggingT (ReaderT LoggingConfig Handler)
type Service api = ServerT api ServiceM
@@ -77,3 +85,7 @@ secondsNow runningDay = do
utcToSeconds :: UTCTime -> Day -> Seconds
utcToSeconds time day =
Seconds $ round $ nominalDiffTimeToSeconds $ diffUTCTime time (UTCTime day 0)
+
+getTzseries :: ServerConfig -> Text -> IO TimeZoneSeries
+getTzseries settings tzname = getTimeZoneSeriesFromOlsonFile
+ (serverConfigZoneinfoPath settings </> T.unpack tzname)