From c13f01a012f80ef578eb10715952e84e7a3eb057 Mon Sep 17 00:00:00 2001 From: Mats Rauhala Date: Fri, 28 Mar 2025 19:55:59 +0200 Subject: [PATCH] Service --- default.nix | 16 ++++--- src/AI.hs | 18 +++++--- src/Creds.hs | 12 +++--- src/HTTP.hs | 64 ++++++++++++++++------------ src/M.hs | 23 +++++++--- src/MyLib.hs | 97 ++++++++++++++++++++++++++++++++----------- src/Server.hs | 85 +++++++++++++++++++++++++++++++++++++ src/Wallabag.hs | 66 +++++++++++++++++++++-------- wallabag-tagger.cabal | 6 +++ 9 files changed, 296 insertions(+), 91 deletions(-) create mode 100644 src/Server.hs diff --git a/default.nix b/default.nix index e345047..ce081f3 100644 --- a/default.nix +++ b/default.nix @@ -1,8 +1,10 @@ { mkDerivation, aeson, annotated-exception, base, bytestring , containers, directory, ekg-core, exceptions, filepath -, generic-lens, katip, lens, lens-aeson, lib, monad-control, mtl -, optparse-generic, pandoc, pipes, retry, stm, streamly -, streamly-core, text, time, transformers-base, unliftio, wreq +, generic-lens, http-client, http-types, katip, lens, lens-aeson +, lib, monad-control, mtl, optparse-generic, pandoc, pipes +, pipes-concurrency, retry, servant, servant-server, stm, streamly +, streamly-core, text, time, transformers-base, unliftio +, unordered-containers, wai, warp, wreq }: mkDerivation { pname = "wallabag-tagger"; @@ -12,9 +14,11 @@ mkDerivation { isExecutable = true; libraryHaskellDepends = [ aeson annotated-exception base bytestring containers directory - ekg-core exceptions filepath generic-lens katip lens lens-aeson - monad-control mtl optparse-generic pandoc pipes retry stm streamly - streamly-core text time transformers-base unliftio wreq + ekg-core exceptions filepath generic-lens http-client http-types + katip lens lens-aeson monad-control mtl optparse-generic pandoc + pipes pipes-concurrency retry servant servant-server stm streamly + streamly-core text time transformers-base unliftio + unordered-containers wai warp wreq ]; executableHaskellDepends = [ base ]; testHaskellDepends = [ base ]; diff --git a/src/AI.hs b/src/AI.hs index 0ad2e9b..c2f6906 100644 --- a/src/AI.hs +++ b/src/AI.hs @@ -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 -> diff --git a/src/Creds.hs b/src/Creds.hs index 4ee8c69..874ae1d 100644 --- a/src/Creds.hs +++ b/src/Creds.hs @@ -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 diff --git a/src/HTTP.hs b/src/HTTP.hs index 8dc7635..b70b757 100644 --- a/src/HTTP.hs +++ b/src/HTTP.hs @@ -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) diff --git a/src/M.hs b/src/M.hs index f77ba2d..3885133 100644 --- a/src/M.hs +++ b/src/M.hs @@ -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 diff --git a/src/MyLib.hs b/src/MyLib.hs index 2901824..bfff40d 100644 --- a/src/MyLib.hs +++ b/src/MyLib.hs @@ -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 diff --git a/src/Server.hs b/src/Server.hs new file mode 100644 index 0000000..87fc5cb --- /dev/null +++ b/src/Server.hs @@ -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 diff --git a/src/Wallabag.hs b/src/Wallabag.hs index f7538e9..1eb4b37 100644 --- a/src/Wallabag.hs +++ b/src/Wallabag.hs @@ -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 () diff --git a/wallabag-tagger.cabal b/wallabag-tagger.cabal index 826ecf2..ae1d7f4 100644 --- a/wallabag-tagger.cabal +++ b/wallabag-tagger.cabal @@ -25,6 +25,7 @@ library Wallabag.Auth MyLib HTTP + Server Tracer -- other-modules: -- other-extensions: @@ -37,10 +38,15 @@ library , ekg-core , retry , pipes + , pipes-concurrency , streamly , streamly-core , unordered-containers , monad-control + , servant + , servant-server + , warp + , wai , transformers-base , pandoc , annotated-exception