Service
This commit is contained in:
1 parent
05cf5df18f
commit
c13f01a012
9 files changed
+296
-91
No files matched your search
@@ -10,24 +10,32 @@ import Control.Monad.Reader (MonadReader, liftIO)
|
||||
import Creds
|
||||
import Data.Aeson (object, (.=))
|
||||
import Data.Aeson.Lens (key, _Array, _String)
|
||||
import Data.Functor.Contravariant ((>$<))
|
||||
import Data.Text (Text)
|
||||
import Data.Text.Strict.Lens (utf8)
|
||||
import HTTP (HTTPMetric, Retry (SafeTransient), Trace, postWith, tracerMetrics)
|
||||
import M (HasOpenrouterMetrics, openrouterMetrics)
|
||||
import HTTP (HTTPMetric, Retry (SafeTransient), Trace, logFormatter, postWith, tracerMetrics)
|
||||
import M (HasKatipTracer, HasOpenrouterMetrics, KatipTracer, katipTracer, openrouterMetrics)
|
||||
import Network.Wreq (defaults, header, responseBody)
|
||||
import Tracer (Tracer (..))
|
||||
import UnliftIO (MonadIO)
|
||||
|
||||
mkTracerMetrics :: (MonadReader r m, HasOpenrouterMetrics r HTTPMetric) => m (Tracer HTTP.Trace)
|
||||
mkTracerMetrics ::
|
||||
( MonadReader r m,
|
||||
HasOpenrouterMetrics r HTTPMetric,
|
||||
HasKatipTracer r KatipTracer
|
||||
) =>
|
||||
m (Tracer HTTP.Trace)
|
||||
mkTracerMetrics = do
|
||||
l <- view katipTracer
|
||||
m <- view openrouterMetrics
|
||||
pure $ HTTP.tracerMetrics m
|
||||
pure $ (logFormatter >$< l) <> HTTP.tracerMetrics m
|
||||
|
||||
completions ::
|
||||
( MonadIO m,
|
||||
MonadCatch m,
|
||||
MonadReader r0 m,
|
||||
HasOpenrouterMetrics r0 HTTPMetric
|
||||
HasOpenrouterMetrics r0 HTTPMetric,
|
||||
HasKatipTracer r0 KatipTracer
|
||||
) =>
|
||||
Creds ->
|
||||
Text ->
|
||||
|
||||
+7
-5
@@ -30,11 +30,13 @@ instance FromJSON Creds where
|
||||
|
||||
makeFields ''Creds
|
||||
|
||||
|
||||
getCreds :: IO Creds
|
||||
getCreds = do
|
||||
home <- getHomeDirectory
|
||||
let path = home </> ".config/wallabag/creds.json"
|
||||
readCreds :: FilePath -> IO Creds
|
||||
readCreds path = do
|
||||
c <- eitherDecode <$> LBS.readFile path
|
||||
either error pure c
|
||||
|
||||
getDefaultCreds :: IO Creds
|
||||
getDefaultCreds = do
|
||||
home <- getHomeDirectory
|
||||
let path = home </> ".config/wallabag/creds.json"
|
||||
readCreds path
|
||||
+37
-27
@@ -7,8 +7,9 @@ module HTTP
|
||||
getWith,
|
||||
Retry (..),
|
||||
Trace (..),
|
||||
HTTPMetric(..),
|
||||
tracerMetrics
|
||||
HTTPMetric (..),
|
||||
tracerMetrics,
|
||||
logFormatter,
|
||||
)
|
||||
where
|
||||
|
||||
@@ -16,21 +17,24 @@ import Control.Exception.Annotated (catch, throw)
|
||||
import Control.Monad.Catch (Handler (Handler), SomeException)
|
||||
import Control.Retry (RetryStatus (RetryStatus, rsIterNumber), recovering, retryPolicyDefault)
|
||||
import Data.ByteString.Lazy (ByteString)
|
||||
import Data.Text qualified as T
|
||||
import Data.Time.Clock.POSIX (POSIXTime, getPOSIXTime)
|
||||
import Katip qualified as K
|
||||
import Network.HTTP.Client (HttpException (..), HttpExceptionContent (..), Response (..), method)
|
||||
import Network.HTTP.Types (methodGet, methodHead, status408, status429, status500, status502, status503, status504)
|
||||
import Network.Wreq qualified as Wreq
|
||||
import Network.Wreq.Types qualified as Wreq
|
||||
import Tracer
|
||||
import System.Metrics.Distribution (Distribution)
|
||||
import System.Metrics.Counter (Counter)
|
||||
import qualified System.Metrics.Counter as Counter
|
||||
import qualified System.Metrics.Distribution as Distribution
|
||||
import System.Metrics.Counter qualified as Counter
|
||||
import System.Metrics.Distribution (Distribution)
|
||||
import System.Metrics.Distribution qualified as Distribution
|
||||
import Text.Printf (printf)
|
||||
import Tracer
|
||||
|
||||
data Trace
|
||||
= Retrying Int
|
||||
| Failing SomeException
|
||||
| Success POSIXTime POSIXTime
|
||||
= Retrying String Int
|
||||
| Failing String SomeException
|
||||
| Success String POSIXTime POSIXTime
|
||||
deriving (Show)
|
||||
|
||||
data Retry = SafeTransient | Transient | NoRetry
|
||||
@@ -56,9 +60,9 @@ retryRules rules =
|
||||
|| responseStatus res == status502
|
||||
|| responseStatus res == status504
|
||||
|
||||
traceRetry :: Tracer Trace -> RetryStatus -> IO ()
|
||||
traceRetry tracer RetryStatus {rsIterNumber = n} | n > 0 = trace tracer $ Retrying n
|
||||
traceRetry _wtracer _rs = pure ()
|
||||
traceRetry :: String -> Tracer Trace -> RetryStatus -> IO ()
|
||||
traceRetry url tracer RetryStatus {rsIterNumber = n} | n > 0 = trace tracer $ Retrying url n
|
||||
traceRetry _url _wtracer _rs = pure ()
|
||||
|
||||
post :: (Wreq.Postable a) => Tracer Trace -> Retry -> String -> a -> IO (Wreq.Response ByteString)
|
||||
post tracer rules url postable = do
|
||||
@@ -67,10 +71,10 @@ post tracer rules url postable = do
|
||||
recovering
|
||||
retryPolicyDefault
|
||||
(retryRules rules)
|
||||
(\rs -> traceRetry tracer rs >> Wreq.post url postable)
|
||||
`catch` (\e -> trace tracer (Failing e) >> throw e)
|
||||
(\rs -> traceRetry url tracer rs >> Wreq.post url postable)
|
||||
`catch` (\e -> trace tracer (Failing url e) >> throw e)
|
||||
end <- getPOSIXTime
|
||||
trace tracer $ Success start end
|
||||
trace tracer $ Success url start end
|
||||
pure res
|
||||
|
||||
postWith :: (Wreq.Postable a) => Tracer Trace -> Retry -> Wreq.Options -> String -> a -> IO (Wreq.Response ByteString)
|
||||
@@ -80,10 +84,10 @@ postWith tracer rules opts url postable = do
|
||||
recovering
|
||||
retryPolicyDefault
|
||||
(retryRules rules)
|
||||
(\rs -> traceRetry tracer rs >> Wreq.postWith opts url postable)
|
||||
`catch` (\e -> trace tracer (Failing e) >> throw e)
|
||||
(\rs -> traceRetry url tracer rs >> Wreq.postWith opts url postable)
|
||||
`catch` (\e -> trace tracer (Failing url e) >> throw e)
|
||||
end <- getPOSIXTime
|
||||
trace tracer $ Success start end
|
||||
trace tracer $ Success url start end
|
||||
pure res
|
||||
|
||||
get :: Tracer Trace -> Retry -> String -> IO (Response ByteString)
|
||||
@@ -93,10 +97,10 @@ get tracer rules url = do
|
||||
recovering
|
||||
retryPolicyDefault
|
||||
(retryRules rules)
|
||||
(\rs -> traceRetry tracer rs >> Wreq.get url)
|
||||
`catch` (\e -> trace tracer (Failing e) >> throw e)
|
||||
(\rs -> traceRetry url tracer rs >> Wreq.get url)
|
||||
`catch` (\e -> trace tracer (Failing url e) >> throw e)
|
||||
end <- getPOSIXTime
|
||||
trace tracer $ Success start end
|
||||
trace tracer $ Success url start end
|
||||
pure res
|
||||
|
||||
getWith :: Tracer Trace -> Retry -> Wreq.Options -> String -> IO (Response ByteString)
|
||||
@@ -106,10 +110,10 @@ getWith tracer rules opts url = do
|
||||
recovering
|
||||
retryPolicyDefault
|
||||
(retryRules rules)
|
||||
(\rs -> traceRetry tracer rs >> Wreq.getWith opts url)
|
||||
`catch` (\e -> trace tracer (Failing e) >> throw e)
|
||||
(\rs -> traceRetry url tracer rs >> Wreq.getWith opts url)
|
||||
`catch` (\e -> trace tracer (Failing url e) >> throw e)
|
||||
end <- getPOSIXTime
|
||||
trace tracer $ Success start end
|
||||
trace tracer $ Success url start end
|
||||
pure res
|
||||
|
||||
data HTTPMetric = HTTPMetric
|
||||
@@ -120,6 +124,12 @@ data HTTPMetric = HTTPMetric
|
||||
|
||||
tracerMetrics :: HTTPMetric -> Tracer HTTP.Trace
|
||||
tracerMetrics m = Tracer $ \case
|
||||
HTTP.Retrying _ -> Counter.inc (httpRetries m)
|
||||
HTTP.Failing _ -> Counter.inc (httpFailures m)
|
||||
HTTP.Success start end -> Distribution.add (httpRequests m) (realToFrac end - realToFrac start)
|
||||
Retrying _ _ -> Counter.inc (httpRetries m)
|
||||
Failing _ _ -> Counter.inc (httpFailures m)
|
||||
Success _ start end -> Distribution.add (httpRequests m) (realToFrac end - realToFrac start)
|
||||
|
||||
logFormatter :: Trace -> Maybe (K.Severity, K.LogStr)
|
||||
logFormatter = \case
|
||||
Retrying url n -> Just (K.WarningS, K.ls $ T.pack $ printf "Retrying (%s) (%d)" url n)
|
||||
Failing _ _ -> Nothing -- The error will propagate up, no point in logging it twice
|
||||
Success url _ _ -> Just (K.DebugS, K.ls $ T.pack $ printf "Fetching (%s)" url)
|
||||
@@ -3,6 +3,7 @@
|
||||
{-# LANGUAGE TemplateHaskell #-}
|
||||
{-# LANGUAGE UndecidableInstances #-}
|
||||
{-# OPTIONS_GHC -Wall #-}
|
||||
{-# LANGUAGE LambdaCase #-}
|
||||
|
||||
module M where
|
||||
|
||||
@@ -23,6 +24,9 @@ import System.Metrics qualified as EKG
|
||||
import UnliftIO (MonadUnliftIO)
|
||||
import Wallabag.Auth (Auth, Auth' (..))
|
||||
import qualified Data.HashMap.Strict as HM
|
||||
import Tracer (Tracer(..))
|
||||
|
||||
type KatipTracer = Tracer (Maybe (K.Severity, K.LogStr))
|
||||
|
||||
data S = S
|
||||
{ _sTokenCache :: Auth,
|
||||
@@ -31,7 +35,9 @@ data S = S
|
||||
_sLogContext :: K.LogContexts,
|
||||
_sCreds :: Creds,
|
||||
_sWallabagMetrics :: HTTPMetric,
|
||||
_sOpenrouterMetrics :: HTTPMetric
|
||||
_sOpenrouterMetrics :: HTTPMetric,
|
||||
_sKatipTracer :: KatipTracer,
|
||||
_sMetricsStore :: EKG.Store
|
||||
}
|
||||
deriving (Generic)
|
||||
|
||||
@@ -64,8 +70,13 @@ instance K.KatipContext M where
|
||||
getKatipNamespace = view logNamespace
|
||||
localKatipNamespace = local . over logNamespace
|
||||
|
||||
initialS :: K.Severity -> (S -> IO a) -> IO a
|
||||
initialS severity f = do
|
||||
mkKatipTrace :: K.LogEnv -> K.Namespace -> KatipTracer
|
||||
mkKatipTrace le ns = Tracer $ \case
|
||||
Nothing -> pure ()
|
||||
Just (severity, msg) -> K.runKatipContextT le () ns $ K.logFM severity msg
|
||||
|
||||
initialS :: Maybe FilePath -> K.Severity -> (S -> IO a) -> IO a
|
||||
initialS credsPath severity f = do
|
||||
store <- EKG.newStore
|
||||
wbMetrics <-
|
||||
HTTPMetric
|
||||
@@ -77,12 +88,14 @@ initialS severity f = do
|
||||
<$> EKG.createDistribution "openrouter.requests" store
|
||||
<*> EKG.createCounter "openrouter.retries" store
|
||||
<*> EKG.createCounter "openrouter.failures" store
|
||||
c <- getCreds
|
||||
c <- maybe getDefaultCreds readCreds credsPath
|
||||
handleScribe <- K.mkHandleScribe K.ColorIfTerminal stdout (K.permitItem severity) K.V2
|
||||
let makeLogEnv = K.registerScribe "stdout" handleScribe K.defaultScribeSettings =<< K.initLogEnv "wallabag" "production"
|
||||
res <- bracket makeLogEnv K.closeScribes $ \le -> do
|
||||
let kTrace = mkKatipTrace le ns
|
||||
ns = "main"
|
||||
cache <- newTVarIO Unauthenticated
|
||||
f (S cache "main" le mempty c wbMetrics orMetrics)
|
||||
f (S cache ns le mempty c wbMetrics orMetrics kTrace store)
|
||||
-- TODO: This is just for debug purposes
|
||||
EKG.sampleAll store >>= mapM_ print . HM.toList
|
||||
pure res
|
||||
+72
-25
@@ -5,7 +5,9 @@
|
||||
module MyLib (defaultMain, test) where
|
||||
|
||||
import AI
|
||||
import Control.Exception.Annotated.UnliftIO (Exception, SomeException, catch, throw)
|
||||
import Control.Lens (over, re, strict, view, (&), (.~), (^.))
|
||||
import Control.Monad (forever)
|
||||
import Data.Aeson (decode, encode)
|
||||
import Data.Either (fromRight)
|
||||
import Data.Maybe (fromMaybe)
|
||||
@@ -15,16 +17,27 @@ import Data.Text qualified as T
|
||||
import Data.Text.Strict.Lens (utf8)
|
||||
import Katip qualified as K
|
||||
import M
|
||||
import Options.Generic (Generic, ParseRecord (..), getRecord, unHelpful, type (<?>))
|
||||
import Streamly.Data.Fold.Prelude qualified as Fold
|
||||
import Streamly.Data.Stream.Prelude qualified as Stream
|
||||
import Options.Generic (Generic, ParseRecord (..), getRecord, unDefValue, unHelpful, type (<!>), type (<?>))
|
||||
import Pipes (Pipe, for, lift, runEffect, (>->))
|
||||
import Pipes.Prelude qualified as P
|
||||
import Text.Pandoc (PandocError, def, readHtml, runPure, writeMarkdown)
|
||||
import UnliftIO (MonadUnliftIO, liftIO, atomicModifyIORef')
|
||||
import UnliftIO.Async (mapConcurrently_)
|
||||
import UnliftIO.Concurrent (threadDelay)
|
||||
import UnliftIO.Exception (displayException, isAsyncException)
|
||||
import Wallabag
|
||||
import Network.Wai.Handler.Warp (setPort, setLogger, defaultSettings, runSettings)
|
||||
import Server (app, Health (..))
|
||||
import UnliftIO.IORef (newIORef, IORef, writeIORef)
|
||||
|
||||
data Options = Options
|
||||
{ full :: Bool <?> "Go through all the entries",
|
||||
verbose :: Bool
|
||||
}
|
||||
data Options
|
||||
= Quick {verbose :: Bool}
|
||||
| Full {verbose :: Bool}
|
||||
| Daemon
|
||||
{ verbose :: Bool,
|
||||
credentials :: FilePath <?> "Path to the credentials.json",
|
||||
port :: Int <?> "Port to run the diagnostics" <!> "8094"
|
||||
}
|
||||
deriving (Generic, Show)
|
||||
|
||||
instance ParseRecord Options
|
||||
@@ -41,22 +54,57 @@ htmlToMarkdown input = runPure $ do
|
||||
doc <- readHtml def input
|
||||
writeMarkdown def doc
|
||||
|
||||
processableEntries :: (Monad m) => Stream.Stream m Entry -> Stream.Stream m Entry
|
||||
processableEntries = Stream.filter (not . tooShort) . Stream.filter missingTags
|
||||
processableEntries :: (Monad m) => Pipe Entry Entry m ()
|
||||
processableEntries = P.filter (not . tooShort) >-> P.filter missingTags
|
||||
|
||||
markdownEntry :: (Monad m) => Stream.Stream m Entry -> Stream.Stream m Entry
|
||||
markdownEntry = fmap (over content (\x -> fromRight x $ htmlToMarkdown x))
|
||||
-- Convert html into markdown
|
||||
markdownEntry :: (Monad m) => Pipe Entry Entry m ()
|
||||
markdownEntry = P.map (over content (\x -> fromRight x $ htmlToMarkdown x))
|
||||
|
||||
run :: K.Severity -> Scan -> IO ()
|
||||
run sev scan = initialS sev $ \s -> do
|
||||
runM s $ do
|
||||
Stream.fold Fold.drain $
|
||||
sourceEntries scan
|
||||
& markdownEntry & processableEntries
|
||||
& Stream.parMapM (Stream.maxThreads 5) doWork
|
||||
run sev scan = initialS Nothing sev $ \s -> do
|
||||
runM s $ runScan scan
|
||||
|
||||
runScan :: Scan -> M ()
|
||||
runScan scan =
|
||||
Pipes.runEffect $
|
||||
Pipes.for
|
||||
(sourceEntries scan >-> markdownEntry >-> processableEntries)
|
||||
(lift . doWork)
|
||||
|
||||
runDaemon :: Int -> FilePath -> K.Severity -> IO ()
|
||||
runDaemon diagnosticsPort credsPath sev = initialS (Just credsPath) sev $ \s -> runM s $ do
|
||||
let serverSettings = setPort diagnosticsPort defaultSettings
|
||||
health <- newIORef OK
|
||||
|
||||
mapConcurrently_
|
||||
id
|
||||
[ backgroundRunner hourly $ worker health Unarchived,
|
||||
backgroundRunner weekly $ worker health Everything,
|
||||
worker health Unarchived, -- Do a quick scan on startup
|
||||
liftIO $ runSettings serverSettings (app s health)
|
||||
]
|
||||
K.logFM K.InfoS "finished"
|
||||
where
|
||||
worker health scan = handle (reportError health) $ do
|
||||
runScan scan
|
||||
writeIORef health OK
|
||||
weekly = 7 * 86400 * 1_000_000
|
||||
hourly = 60 * 60 * 1_000_000
|
||||
backgroundRunner delay action = forever $ do
|
||||
threadDelay delay
|
||||
action
|
||||
reportError :: (K.KatipContext m, K.Katip m, MonadUnliftIO m) => IORef Health -> SomeException -> m ()
|
||||
reportError health e
|
||||
| isAsyncException e = throw e
|
||||
| otherwise = do
|
||||
writeIORef health Degraded
|
||||
K.logFM K.ErrorS $ K.ls $ displayException e
|
||||
handle :: (Exception e, MonadUnliftIO m) => (e -> m a) -> m a -> m a
|
||||
handle = flip catch
|
||||
|
||||
test :: IO ()
|
||||
test = run K.DebugS Quick
|
||||
test = run K.DebugS Unarchived
|
||||
|
||||
suggestTags :: Text -> M [Text]
|
||||
suggestTags txt = do
|
||||
@@ -111,15 +159,14 @@ hardLabels =
|
||||
"AI"
|
||||
]
|
||||
|
||||
|
||||
|
||||
defaultMain :: IO ()
|
||||
defaultMain = do
|
||||
opts <- getRecord "Wallabag tagger"
|
||||
let sev = severity opts
|
||||
if unHelpful $ full opts
|
||||
then run sev Full
|
||||
else run sev Quick
|
||||
case opts of
|
||||
Quick {verbose} -> run (severity verbose) Unarchived
|
||||
Full {verbose} -> run (severity verbose) Everything
|
||||
Daemon {verbose, credentials, port} ->
|
||||
runDaemon (unHelpful $ unDefValue port) (unHelpful credentials) (severity verbose)
|
||||
where
|
||||
severity Options {verbose = True} = K.DebugS
|
||||
severity True = K.DebugS
|
||||
severity _ = K.InfoS
|
||||
@@ -0,0 +1,85 @@
|
||||
{-# LANGUAGE DataKinds #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
|
||||
module Server where
|
||||
|
||||
import Control.Exception.Annotated (try)
|
||||
import Control.Lens (lazy, re, view, _Show)
|
||||
import Control.Monad.Except (ExceptT (..))
|
||||
import Data.Aeson (FromJSON, ToJSON, object, toJSON, (.=))
|
||||
import Data.HashMap.Strict (HashMap)
|
||||
import Data.IORef (IORef)
|
||||
import Data.Text (Text)
|
||||
import Data.Text.Strict.Lens (packed, utf8)
|
||||
import GHC.Generics (Generic)
|
||||
import M
|
||||
import Servant
|
||||
import Servant.Server.Generic
|
||||
import System.Metrics (Value)
|
||||
import System.Metrics qualified as EKG
|
||||
import System.Metrics.Distribution qualified as D
|
||||
import UnliftIO (liftIO)
|
||||
import UnliftIO.IORef (readIORef)
|
||||
import Data.Coerce (coerce)
|
||||
|
||||
data Health
|
||||
= OK
|
||||
| Degraded
|
||||
deriving (Show, Read, Generic)
|
||||
|
||||
instance ToJSON Health
|
||||
|
||||
instance FromJSON Health
|
||||
|
||||
instance MimeRender PlainText Health where
|
||||
mimeRender _ = view (re _Show . packed . re utf8 . lazy)
|
||||
|
||||
newtype SampleValue = SampleValue Value
|
||||
|
||||
instance ToJSON SampleValue where
|
||||
toJSON (SampleValue v) =
|
||||
case v of
|
||||
EKG.Counter n -> object ["counter" .= n]
|
||||
EKG.Gauge n -> object ["gauge" .= n]
|
||||
EKG.Label l -> object ["label" .= l]
|
||||
EKG.Distribution stats ->
|
||||
object
|
||||
[ "mean" .= D.mean stats,
|
||||
"variance" .= D.variance stats,
|
||||
"count" .= D.count stats,
|
||||
"sum" .= D.sum stats,
|
||||
"min" .= D.min stats,
|
||||
"max" .= D.max stats
|
||||
]
|
||||
|
||||
newtype Metrics = Metrics (HashMap Text SampleValue)
|
||||
deriving (ToJSON)
|
||||
|
||||
data WallabagRoutes mode = WallabagRoutes
|
||||
{ _getHealth :: mode :- "health" :> Get '[PlainText, JSON] Health,
|
||||
_getMetrics :: mode :- "metrics" :> Get '[JSON] Metrics
|
||||
}
|
||||
deriving (Generic)
|
||||
|
||||
-- I'm having the IORef Health outside of the state because it doesn't
|
||||
-- fit into the general environment I think. I might change my mind later
|
||||
wallabagServer :: IORef Health -> WallabagRoutes (AsServerT M)
|
||||
wallabagServer health =
|
||||
WallabagRoutes
|
||||
{ _getHealth = getHealth health,
|
||||
_getMetrics = getMetrics
|
||||
}
|
||||
|
||||
getHealth :: IORef Health -> M Health
|
||||
getHealth = readIORef
|
||||
|
||||
getMetrics :: M Metrics
|
||||
getMetrics = do
|
||||
sample <- view metricsStore >>= liftIO . EKG.sampleAll
|
||||
pure $ coerce sample
|
||||
|
||||
app :: S -> IORef Health -> Application
|
||||
app st health = genericServeT nat (wallabagServer health)
|
||||
where
|
||||
nat :: forall x. M x -> Handler x
|
||||
nat f = Handler $ ExceptT . try $ runM st f
|
||||
+48
-18
@@ -8,16 +8,19 @@
|
||||
|
||||
module Wallabag where
|
||||
|
||||
import Control.Concurrent.STM (atomically, readTVar, retry, writeTVar)
|
||||
import Control.Concurrent.STM (readTVar, retry, writeTVar)
|
||||
import Control.Exception (Exception)
|
||||
import Control.Exception.Annotated (Annotation (..), checkpoint, throw)
|
||||
import Control.Exception.Annotated.UnliftIO (catch)
|
||||
import Control.Lens (re, to, traversed, view, (&), (.~), (^.), (^..), (^?))
|
||||
import Control.Lens.TH (makeFields)
|
||||
import Control.Monad ((<=<))
|
||||
import Control.Monad.Catch (MonadCatch, MonadThrow)
|
||||
import Control.Monad.Reader (MonadReader, liftIO)
|
||||
import Creds
|
||||
import Data.Aeson (Value)
|
||||
import Data.Aeson.Lens (key, _Array, _Integer, _String)
|
||||
import Data.Functor.Contravariant ((>$<))
|
||||
import Data.Generics.Product (HasType (..))
|
||||
import Data.Maybe (mapMaybe)
|
||||
import Data.Text (Text)
|
||||
@@ -25,13 +28,17 @@ import Data.Text qualified as T
|
||||
import Data.Text.Strict.Lens (unpacked, utf8)
|
||||
import Data.Time (addUTCTime, getCurrentTime, secondsToNominalDiffTime)
|
||||
import GHC.Stack (HasCallStack)
|
||||
import HTTP (HTTPMetric, Retry (SafeTransient), Trace, getWith, post, postWith, tracerMetrics)
|
||||
import HTTP (HTTPMetric, Retry (SafeTransient), Trace (..), getWith, logFormatter, post, postWith, tracerMetrics)
|
||||
import Katip qualified as K
|
||||
import M (HasWallabagMetrics, wallabagMetrics)
|
||||
import M (HasKatipTracer, HasWallabagMetrics, KatipTracer, katipTracer, wallabagMetrics)
|
||||
import Network.HTTP.Client (HttpException)
|
||||
import Network.Wreq (FormParam (..), defaults, header, responseBody)
|
||||
import Streamly.Data.Stream.Prelude qualified as Stream
|
||||
import Pipes ((>->))
|
||||
import Pipes qualified
|
||||
import Pipes.Prelude qualified as P
|
||||
import Tracer (Tracer (..))
|
||||
import UnliftIO (MonadIO)
|
||||
import UnliftIO (MonadIO, MonadUnliftIO)
|
||||
import UnliftIO.STM (atomically)
|
||||
import Wallabag.Auth
|
||||
|
||||
data Entry = Entry
|
||||
@@ -48,7 +55,8 @@ authenticate ::
|
||||
( MonadReader r m,
|
||||
MonadIO m,
|
||||
K.KatipContext m,
|
||||
HasWallabagMetrics r HTTPMetric
|
||||
HasWallabagMetrics r HTTPMetric,
|
||||
HasKatipTracer r KatipTracer
|
||||
) =>
|
||||
Creds ->
|
||||
m (Maybe Auth')
|
||||
@@ -73,15 +81,19 @@ data WallabagException7g = MissingTokenException deriving (Show)
|
||||
instance Exception WallabagException7g
|
||||
|
||||
getToken ::
|
||||
forall m r.
|
||||
( HasType Creds r,
|
||||
HasType Auth r,
|
||||
MonadReader r m,
|
||||
MonadIO m,
|
||||
K.KatipContext m,
|
||||
HasWallabagMetrics r HTTPMetric
|
||||
HasWallabagMetrics r HTTPMetric,
|
||||
HasKatipTracer r KatipTracer,
|
||||
MonadUnliftIO m,
|
||||
MonadThrow m
|
||||
) =>
|
||||
m (Maybe Text)
|
||||
getToken = do
|
||||
getToken = handle revertToUnauthenticated $ do
|
||||
now <- liftIO getCurrentTime
|
||||
cache <- view (typed @Auth)
|
||||
token <- liftIO $ atomically $ do
|
||||
@@ -94,13 +106,20 @@ getToken = do
|
||||
| otherwise -> Nothing <$ writeTVar cache Authenticating
|
||||
maybe (reAuthenticate cache) (pure . Just) token
|
||||
where
|
||||
handle :: (Exception e) => (e -> m a) -> m a -> m a
|
||||
handle = flip catch
|
||||
revertToUnauthenticated :: HttpException -> m (Maybe Text)
|
||||
revertToUnauthenticated e = do
|
||||
cache <- view (typed @Auth)
|
||||
atomically $ writeTVar cache Unauthenticated
|
||||
throw e
|
||||
reAuthenticate cache = do
|
||||
a <- view (typed @Creds) >>= authenticate
|
||||
liftIO $ case a of
|
||||
Nothing -> Nothing <$ atomically (writeTVar cache Unauthenticated)
|
||||
Just a' -> Just (accessToken a') <$ atomically (writeTVar cache a')
|
||||
|
||||
data Scan = Quick | Full
|
||||
data Scan = Unarchived | Everything
|
||||
|
||||
sourceEntries ::
|
||||
forall r m.
|
||||
@@ -112,16 +131,19 @@ sourceEntries ::
|
||||
MonadThrow m,
|
||||
MonadCatch m,
|
||||
HasCallStack,
|
||||
HasWallabagMetrics r HTTPMetric
|
||||
HasWallabagMetrics r HTTPMetric,
|
||||
HasKatipTracer r KatipTracer,
|
||||
MonadUnliftIO m
|
||||
) =>
|
||||
Scan ->
|
||||
Stream.Stream m Entry
|
||||
sourceEntries s = Stream.unfoldrM go (Just startUrl) & Stream.concatMap (Stream.fromList . mapMaybe toEntry)
|
||||
Pipes.Producer Entry m ()
|
||||
sourceEntries s = P.unfoldr (toEither <=< go) (Just startUrl) >-> P.mapFoldable (mapMaybe toEntry)
|
||||
where
|
||||
toEither = pure . maybe (Left ()) Right
|
||||
startUrl =
|
||||
case s of
|
||||
Quick -> "https://app.wallabag.it/api/entries.json?archive=0"
|
||||
Full -> "https://app.wallabag.it/api/entries.json"
|
||||
Unarchived -> "https://app.wallabag.it/api/entries.json?archive=0"
|
||||
Everything -> "https://app.wallabag.it/api/entries.json"
|
||||
go :: Maybe String -> m (Maybe ([Value], Maybe String))
|
||||
go Nothing = pure Nothing
|
||||
go (Just url) = checkpoint (Annotation url) $ do
|
||||
@@ -141,10 +163,16 @@ sourceEntries s = Stream.unfoldrM go (Just startUrl) & Stream.concatMap (Stream.
|
||||
<*> Just (v ^.. key "tags" . _Array . traversed . key "label" . _String)
|
||||
<*> v ^? key "title" . _String
|
||||
|
||||
mkTracerMetrics :: (MonadReader r m, HasWallabagMetrics r HTTPMetric) => m (Tracer HTTP.Trace)
|
||||
mkTracerMetrics ::
|
||||
( MonadReader r m,
|
||||
HasWallabagMetrics r HTTPMetric,
|
||||
HasKatipTracer r KatipTracer
|
||||
) =>
|
||||
m (Tracer HTTP.Trace)
|
||||
mkTracerMetrics = do
|
||||
m <- view wallabagMetrics
|
||||
pure $ HTTP.tracerMetrics m
|
||||
l <- view katipTracer
|
||||
metrics <- view wallabagMetrics
|
||||
pure $ (HTTP.logFormatter >$< l) <> HTTP.tracerMetrics metrics
|
||||
|
||||
applyTags ::
|
||||
( HasType Auth r,
|
||||
@@ -155,7 +183,9 @@ applyTags ::
|
||||
K.KatipContext m,
|
||||
MonadThrow m,
|
||||
MonadCatch m,
|
||||
HasWallabagMetrics r HTTPMetric
|
||||
HasWallabagMetrics r HTTPMetric,
|
||||
HasKatipTracer r KatipTracer,
|
||||
MonadUnliftIO m
|
||||
) =>
|
||||
Entry ->
|
||||
m ()
|
||||
|
||||
Reference in new issue
Block a user