{-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE ExplicitNamespaces #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedLists #-} {-# LANGUAGE PartialTypeSignatures #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE TypeApplications #-} -- Implementation of the API. This module is the main point of the program. module Server (application) where import Control.Monad (forever, unless, void, when) import Control.Monad.Extra (maybeM, unlessM, whenM, ifM) import Control.Monad.IO.Class (MonadIO (liftIO)) import Control.Monad.Logger (logWarnN, LoggingT) import Control.Monad.Reader (forM) import Control.Monad.Trans (lift) 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.Time (NominalDiffTime, UTCTime (utctDay), addUTCTime, diffUTCTime, getCurrentTime, nominalDay) import qualified Data.Vector as V import Database.Persist import Database.Persist.Postgresql (SqlBackend, runMigration) import Fmt ((+|), (|+)) import qualified Network.WebSockets as WS import Servant (Application, ServerError (errBody), err401, err404, serve, throwError) import Servant.API (NoContent (..), (:<|>) (..)) import Servant.Server (Handler, hoistServer) import Servant.Swagger (toSwagger) import API import GTFS import Persist import Server.ControlRoom import Server.GTFS_RT (gtfsRealtimeServer) import Server.Util (Service, ServiceM, runService, sendErrorMsg) import Yesod (toWaiAppPlain) import Conferer (fetch, mkConfig) import Extrapolation (Extrapolator (..), LinearExtrapolator) import System.IO.Unsafe import Config (ServerConfig) import Data.ByteString (ByteString) application :: GTFS -> Pool SqlBackend -> IO Application application gtfs dbpool = do doMigration dbpool pure $ serve (Proxy @CompleteAPI) $ hoistServer (Proxy @CompleteAPI) runService $ server gtfs dbpool -- databaseMigration :: ConnectionString -> IO () doMigration pool = runSql pool $ -- TODO: before that, check if the uuid module is enabled -- in sql: check if SELECT * FROM pg_extension WHERE extname = 'uuid-ossp'; -- returns an empty list runMigration migrateAll server :: GTFS -> Pool SqlBackend -> Service CompleteAPI server gtfs@GTFS{..} dbpool = handleDebugAPI :<|> (handleStations :<|> handleTimetable :<|> handleTrip :<|> handleRegister :<|> handleTrainPing (throwError err401) :<|> handleWS :<|> handleDebugState :<|> handleDebugTrain :<|> handleDebugRegister :<|> gtfsRealtimeServer gtfs dbpool) :<|> pure (unsafePerformIO (toWaiAppPlain (ControlRoom gtfs dbpool))) where handleStations = pure stations handleTimetable station maybeDay = do -- TODO: resolve "overlay" trips (perhaps just additional CalendarDates?) day <- liftIO $ maybeM (getCurrentTime <&> utctDay) pure (pure maybeDay) pure -- don't send stations ending at this station . M.filter ((==) station . stationId . stopStation . V.last . tripStops) $ tripsOnDay gtfs day handleTrip trip = case M.lookup trip trips of Just res -> pure res Nothing -> throwError err404 handleRegister tripID RegisterJson{..} = do today <- liftIO getCurrentTime <&> utctDay unless (runsOnDay gtfs tripID today) $ sendErrorMsg "this trip does not run today." expires <- liftIO $ getCurrentTime <&> addUTCTime validityPeriod RunningKey token <- runSql dbpool $ insert (Running expires False tripID today Nothing registerAgent) pure token handleDebugRegister tripID day = do expires <- liftIO $ getCurrentTime <&> addUTCTime validityPeriod RunningKey token <- runSql dbpool $ insert (Running expires False tripID day Nothing "debug key") pure token handleTrainPing onError ping = isTokenValid dbpool (coerce $ trainPingToken ping) >>= \case Nothing -> do onError pure NoContent Just running@Running{..} -> do let anchor = extrapolateAnchorFromPing @LinearExtrapolator gtfs running ping -- TODO: are these always inserted in order? runSql dbpool $ do insert ping last <- selectFirst [TrainAnchorTrip ==. runningTrip, TrainAnchorDay ==. runningDay] [Desc TrainAnchorWhen] -- only insert new estimates if they've actually changed anything when (fmap (trainAnchorDelay . entityVal) last /= Just (trainAnchorDelay anchor)) $ void $ insert anchor pure NoContent handleWS conn = do liftIO $ WS.forkPingThread conn 30 forever $ do msg <- liftIO $ WS.receiveData conn case A.eitherDecode msg of Left err -> do logWarnN ("stray websocket message: "+|show msg|+" (could not decode: "+|err|+")") liftIO $ WS.sendClose conn (C8.pack err) Right ping -> -- 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. liftIO $ void $ handleTrainPing (WS.sendClose conn ("" :: ByteString)) ping handleDebugState = do now <- liftIO getCurrentTime runSql dbpool $ do running <- selectList [RunningBlocked ==. False, RunningExpires >=. now] [] pairs <- forM running $ \(Entity token@(RunningKey uuid) _) -> do entities <- selectList [TrainPingToken ==. token] [] pure (uuid, fmap entityVal entities) pure (M.fromList pairs) handleDebugTrain tripId day = do unless (runsOnDay gtfs tripId day) $ sendErrorMsg ("this trip does not run on "+|day|+".") runSql dbpool $ do tokens <- selectList [RunningTrip ==. tripId, RunningDay ==. day] [] pings <- forM tokens $ \(Entity token _) -> do selectList [TrainPingToken ==. token] [] <&> fmap entityVal pure (concat pings) handleDebugAPI = pure $ toSwagger (Proxy @API) -- TODO: proper debug logging for expired tokens isTokenValid :: MonadIO m => Pool SqlBackend -> Token -> m (Maybe Running) isTokenValid dbpool token = runSql dbpool $ get (coerce token) >>= \case Just trip | not (runningBlocked trip) -> do ifM (hasExpired (runningExpires trip)) (pure Nothing) (pure (Just trip)) _ -> pure Nothing hasExpired :: MonadIO m => UTCTime -> m Bool hasExpired limit = do now <- liftIO getCurrentTime pure (now > limit) validityPeriod :: NominalDiffTime validityPeriod = nominalDay