diff --git a/default.nix b/default.nix index 664579f..e345047 100644 --- a/default.nix +++ b/default.nix @@ -1,7 +1,8 @@ { mkDerivation, aeson, annotated-exception, base, bytestring -, containers, directory, exceptions, filepath, generic-lens, katip -, lens, lens-aeson, lib, mtl, optparse-generic, pandoc, pipes, stm -, streamly, streamly-core, text, time, unliftio, wreq +, 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 }: mkDerivation { pname = "wallabag-tagger"; @@ -11,9 +12,9 @@ mkDerivation { isExecutable = true; libraryHaskellDepends = [ aeson annotated-exception base bytestring containers directory - exceptions filepath generic-lens katip lens lens-aeson mtl - optparse-generic pandoc pipes stm streamly streamly-core text time - unliftio wreq + 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 ]; executableHaskellDepends = [ base ]; testHaskellDepends = [ base ]; diff --git a/src/AI.hs b/src/AI.hs index 3ab0a0b..0ad2e9b 100644 --- a/src/AI.hs +++ b/src/AI.hs @@ -3,19 +3,35 @@ module AI where -import Control.Lens (re, traversed, (&), (.~), (^.), (^?)) -import Control.Monad.Reader (liftIO) +import Control.Exception.Annotated (checkpointCallStack) +import Control.Lens (re, traversed, view, (&), (.~), (^.), (^?)) +import Control.Monad.Catch (MonadCatch) +import Control.Monad.Reader (MonadReader, liftIO) import Creds import Data.Aeson (object, (.=)) import Data.Aeson.Lens (key, _Array, _String) import Data.Text (Text) import Data.Text.Strict.Lens (utf8) -import Network.Wreq (defaults, header, postWith, responseBody) +import HTTP (HTTPMetric, Retry (SafeTransient), Trace, postWith, tracerMetrics) +import M (HasOpenrouterMetrics, openrouterMetrics) +import Network.Wreq (defaults, header, responseBody) +import Tracer (Tracer (..)) import UnliftIO (MonadIO) -import Control.Exception.Annotated (checkpointCallStack) -import Control.Monad.Catch (MonadCatch) -completions :: (MonadIO m, MonadCatch m) => Creds -> Text -> m (Maybe Text) +mkTracerMetrics :: (MonadReader r m, HasOpenrouterMetrics r HTTPMetric) => m (Tracer HTTP.Trace) +mkTracerMetrics = do + m <- view openrouterMetrics + pure $ HTTP.tracerMetrics m + +completions :: + ( MonadIO m, + MonadCatch m, + MonadReader r0 m, + HasOpenrouterMetrics r0 HTTPMetric + ) => + Creds -> + Text -> + m (Maybe Text) completions c prompt = checkpointCallStack $ do let opts = defaults & header "Authorization" .~ ["Bearer " <> c ^. openrouterKey . re utf8] let j = @@ -24,5 +40,6 @@ completions c prompt = checkpointCallStack $ do "messages" .= [object ["role" .= ("user" :: Text), "content" .= prompt]] ] - r <- liftIO $ postWith opts "https://openrouter.ai/api/v1/chat/completions" j + metrics <- mkTracerMetrics + r <- liftIO $ postWith metrics SafeTransient opts "https://openrouter.ai/api/v1/chat/completions" j pure $ r ^? responseBody . key "choices" . _Array . traversed . key "message" . key "content" . _String diff --git a/src/HTTP.hs b/src/HTTP.hs new file mode 100644 index 0000000..8dc7635 --- /dev/null +++ b/src/HTTP.hs @@ -0,0 +1,125 @@ +{-# LANGUAGE LambdaCase #-} + +module HTTP + ( post, + postWith, + get, + getWith, + Retry (..), + Trace (..), + HTTPMetric(..), + tracerMetrics + ) +where + +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.Time.Clock.POSIX (POSIXTime, getPOSIXTime) +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 + +data Trace + = Retrying Int + | Failing SomeException + | Success POSIXTime POSIXTime + deriving (Show) + +data Retry = SafeTransient | Transient | NoRetry + deriving (Eq) + +retryRules :: Retry -> [RetryStatus -> Handler IO Bool] +retryRules NoRetry = [] +retryRules rules = + [ const $ Handler $ \case + HttpExceptionRequest req ConnectionClosed -> pure $ safe req + HttpExceptionRequest req ConnectionTimeout -> pure $ safe req + HttpExceptionRequest req ResponseTimeout -> pure $ safe req + HttpExceptionRequest req (StatusCodeException res _) -> pure (safe req && retryStatus res) + _ -> pure False + ] + where + safe req = rules == SafeTransient || method req == methodGet || method req == methodHead + retryStatus res = + responseStatus res == status503 + || responseStatus res == status408 + || responseStatus res == status429 + || responseStatus res == status500 + || 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 () + +post :: (Wreq.Postable a) => Tracer Trace -> Retry -> String -> a -> IO (Wreq.Response ByteString) +post tracer rules url postable = do + start <- getPOSIXTime + res <- + recovering + retryPolicyDefault + (retryRules rules) + (\rs -> traceRetry tracer rs >> Wreq.post url postable) + `catch` (\e -> trace tracer (Failing e) >> throw e) + end <- getPOSIXTime + trace tracer $ Success start end + pure res + +postWith :: (Wreq.Postable a) => Tracer Trace -> Retry -> Wreq.Options -> String -> a -> IO (Wreq.Response ByteString) +postWith tracer rules opts url postable = do + start <- getPOSIXTime + res <- + recovering + retryPolicyDefault + (retryRules rules) + (\rs -> traceRetry tracer rs >> Wreq.postWith opts url postable) + `catch` (\e -> trace tracer (Failing e) >> throw e) + end <- getPOSIXTime + trace tracer $ Success start end + pure res + +get :: Tracer Trace -> Retry -> String -> IO (Response ByteString) +get tracer rules url = do + start <- getPOSIXTime + res <- + recovering + retryPolicyDefault + (retryRules rules) + (\rs -> traceRetry tracer rs >> Wreq.get url) + `catch` (\e -> trace tracer (Failing e) >> throw e) + end <- getPOSIXTime + trace tracer $ Success start end + pure res + +getWith :: Tracer Trace -> Retry -> Wreq.Options -> String -> IO (Response ByteString) +getWith tracer rules opts url = do + start <- getPOSIXTime + res <- + recovering + retryPolicyDefault + (retryRules rules) + (\rs -> traceRetry tracer rs >> Wreq.getWith opts url) + `catch` (\e -> trace tracer (Failing e) >> throw e) + end <- getPOSIXTime + trace tracer $ Success start end + pure res + +data HTTPMetric = HTTPMetric + { httpRequests :: Distribution, + httpRetries :: Counter, + httpFailures :: Counter + } + +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) diff --git a/src/M.hs b/src/M.hs index 9ab06fc..f77ba2d 100644 --- a/src/M.hs +++ b/src/M.hs @@ -1,8 +1,8 @@ {-# LANGUAGE FunctionalDependencies #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TemplateHaskell #-} -{-# OPTIONS_GHC -Wall #-} {-# LANGUAGE UndecidableInstances #-} +{-# OPTIONS_GHC -Wall #-} module M where @@ -10,23 +10,28 @@ import Control.Concurrent.STM import Control.Exception (bracket) import Control.Lens (over, view) import Control.Lens.TH (makeFields) +import Control.Monad.Base (MonadBase) import Control.Monad.Catch (MonadCatch, MonadThrow) import Control.Monad.Reader (MonadIO, MonadReader, ReaderT, local, runReaderT) +import Control.Monad.Trans.Control (MonadBaseControl) import Creds import GHC.Generics (Generic) +import HTTP (HTTPMetric (HTTPMetric)) import Katip qualified as K import System.IO (stdout) +import System.Metrics qualified as EKG import UnliftIO (MonadUnliftIO) -import Wallabag (Auth, Auth' (..)) -import Control.Monad.Trans.Control (MonadBaseControl) -import Control.Monad.Base (MonadBase) +import Wallabag.Auth (Auth, Auth' (..)) +import qualified Data.HashMap.Strict as HM data S = S { _sTokenCache :: Auth, _sLogNamespace :: K.Namespace, _sLogEnv :: K.LogEnv, _sLogContext :: K.LogContexts, - _sCreds :: Creds + _sCreds :: Creds, + _sWallabagMetrics :: HTTPMetric, + _sOpenrouterMetrics :: HTTPMetric } deriving (Generic) @@ -61,9 +66,23 @@ instance K.KatipContext M where initialS :: K.Severity -> (S -> IO a) -> IO a initialS severity f = do + store <- EKG.newStore + wbMetrics <- + HTTPMetric + <$> EKG.createDistribution "wallabag.requests" store + <*> EKG.createCounter "wallabag.retries" store + <*> EKG.createCounter "wallabag.failures" store + orMetrics <- + HTTPMetric + <$> EKG.createDistribution "openrouter.requests" store + <*> EKG.createCounter "openrouter.retries" store + <*> EKG.createCounter "openrouter.failures" store c <- getCreds handleScribe <- K.mkHandleScribe K.ColorIfTerminal stdout (K.permitItem severity) K.V2 let makeLogEnv = K.registerScribe "stdout" handleScribe K.defaultScribeSettings =<< K.initLogEnv "wallabag" "production" - bracket makeLogEnv K.closeScribes $ \le -> do + res <- bracket makeLogEnv K.closeScribes $ \le -> do cache <- newTVarIO Unauthenticated - f (S cache "main" le mempty c) + f (S cache "main" le mempty c wbMetrics orMetrics) + -- 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 03c39a0..2901824 100644 --- a/src/MyLib.hs +++ b/src/MyLib.hs @@ -16,10 +16,10 @@ 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 Text.Pandoc (PandocError, def, readHtml, runPure, writeMarkdown) import Wallabag -import qualified Streamly.Data.Fold.Prelude as Fold -import qualified Streamly.Data.Stream.Prelude as Stream data Options = Options { full :: Bool "Go through all the entries", @@ -51,8 +51,9 @@ 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 + sourceEntries scan + & markdownEntry & processableEntries + & Stream.parMapM (Stream.maxThreads 5) doWork test :: IO () test = run K.DebugS Quick @@ -110,6 +111,8 @@ hardLabels = "AI" ] + + defaultMain :: IO () defaultMain = do opts <- getRecord "Wallabag tagger" diff --git a/src/Tracer.hs b/src/Tracer.hs new file mode 100644 index 0000000..58e5486 --- /dev/null +++ b/src/Tracer.hs @@ -0,0 +1,9 @@ +{-# LANGUAGE DerivingVia #-} +module Tracer where +import Data.Functor.Contravariant + + +newtype Tracer a = Tracer { trace :: a -> IO () } + deriving Contravariant via Op (IO ()) + deriving (Semigroup, Monoid) via (a -> IO ()) + diff --git a/src/Wallabag.hs b/src/Wallabag.hs index ae6c9e8..f7538e9 100644 --- a/src/Wallabag.hs +++ b/src/Wallabag.hs @@ -1,44 +1,38 @@ {-# LANGUAGE FunctionalDependencies #-} -{-# LANGUAGE StrictData #-} +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StrictData #-} {-# LANGUAGE TemplateHaskell #-} {-# OPTIONS_GHC -Wall #-} module Wallabag where -import Control.Concurrent.STM (TVar, atomically, readTVar, retry, writeTVar) +import Control.Concurrent.STM (atomically, readTVar, retry, writeTVar) import Control.Exception (Exception) +import Control.Exception.Annotated (Annotation (..), checkpoint, throw) import Control.Lens (re, to, traversed, view, (&), (.~), (^.), (^..), (^?)) import Control.Lens.TH (makeFields) -import Control.Monad.Catch (MonadThrow, MonadCatch) -import Control.Exception.Annotated (throw, checkpoint, Annotation (..)) +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.Generics.Product (HasType (..)) +import Data.Maybe (mapMaybe) import Data.Text (Text) import Data.Text qualified as T import Data.Text.Strict.Lens (unpacked, utf8) -import Data.Time (UTCTime, addUTCTime, getCurrentTime, secondsToNominalDiffTime) -import Katip qualified as K -import Network.Wreq (FormParam (..), defaults, getWith, header, post, postWith, responseBody) -import UnliftIO (MonadIO) +import Data.Time (addUTCTime, getCurrentTime, secondsToNominalDiffTime) import GHC.Stack (HasCallStack) -import qualified Streamly.Data.Stream.Prelude as Stream -import Data.Maybe (mapMaybe) - -data Auth' - = Unauthenticated - | Authenticating - | Auth - { accessToken :: Text, - expires :: UTCTime - } - deriving (Show) - -type Auth = TVar Auth' +import HTTP (HTTPMetric, Retry (SafeTransient), Trace, getWith, post, postWith, tracerMetrics) +import Katip qualified as K +import M (HasWallabagMetrics, wallabagMetrics) +import Network.Wreq (FormParam (..), defaults, header, responseBody) +import Streamly.Data.Stream.Prelude qualified as Stream +import Tracer (Tracer (..)) +import UnliftIO (MonadIO) +import Wallabag.Auth data Entry = Entry { _entryIdentifier :: Integer, @@ -50,7 +44,14 @@ data Entry = Entry makeFields ''Entry -authenticate :: (MonadReader r m, MonadIO m, K.KatipContext m) => Creds -> m (Maybe Auth') +authenticate :: + ( MonadReader r m, + MonadIO m, + K.KatipContext m, + HasWallabagMetrics r HTTPMetric + ) => + Creds -> + m (Maybe Auth') authenticate c = do K.logFM K.DebugS "Authenticating" let form = @@ -60,7 +61,8 @@ authenticate c = do "username" := c ^. username, "password" := c ^. password ] - r <- liftIO $ post "https://app.wallabag.it/oauth/v2/token" form + metrics <- mkTracerMetrics + r <- liftIO $ post metrics SafeTransient "https://app.wallabag.it/oauth/v2/token" form now <- liftIO getCurrentTime let expiresIn = r ^? responseBody . key "expires_in" . _Integer . to fromIntegral . to secondsToNominalDiffTime expireTime = fmap (`addUTCTime` now) expiresIn @@ -70,7 +72,15 @@ data WallabagException7g = MissingTokenException deriving (Show) instance Exception WallabagException7g -getToken :: (HasType Creds r, HasType Auth r, MonadReader r m, MonadIO m, K.KatipContext m) => m (Maybe Text) +getToken :: + ( HasType Creds r, + HasType Auth r, + MonadReader r m, + MonadIO m, + K.KatipContext m, + HasWallabagMetrics r HTTPMetric + ) => + m (Maybe Text) getToken = do now <- liftIO getCurrentTime cache <- view (typed @Auth) @@ -101,7 +111,8 @@ sourceEntries :: K.KatipContext m, MonadThrow m, MonadCatch m, - HasCallStack + HasCallStack, + HasWallabagMetrics r HTTPMetric ) => Scan -> Stream.Stream m Entry @@ -114,10 +125,11 @@ sourceEntries s = Stream.unfoldrM go (Just startUrl) & Stream.concatMap (Stream. go :: Maybe String -> m (Maybe ([Value], Maybe String)) go Nothing = pure Nothing go (Just url) = checkpoint (Annotation url) $ do + metrics <- mkTracerMetrics t <- maybe (throw MissingTokenException) pure =<< getToken let opts = defaults & header "Authorization" .~ ["Bearer " <> t ^. re utf8] K.logFM K.DebugS $ K.ls ("Fetching entries '" <> url <> "'") - r <- liftIO $ getWith opts url + r <- liftIO $ getWith metrics SafeTransient opts url let items = r ^.. responseBody . key "_embedded" . key "items" . _Array . traversed let next = r ^? responseBody . key "_links" . key "next" . key "href" . _String . unpacked pure $ Just (items, next) @@ -129,6 +141,11 @@ 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 = do + m <- view wallabagMetrics + pure $ HTTP.tracerMetrics m + applyTags :: ( HasType Auth r, HasType Creds r, @@ -137,7 +154,8 @@ applyTags :: HasCallStack, K.KatipContext m, MonadThrow m, - MonadCatch m + MonadCatch m, + HasWallabagMetrics r HTTPMetric ) => Entry -> m () @@ -146,5 +164,6 @@ applyTags e = checkpoint (Annotation $ e ^. identifier) $ do let opts = defaults & header "Authorization" .~ ["Bearer " <> t ^. re utf8] K.logFM K.InfoS (K.ls ("Applying tags for '" <> e ^. title <> "': " <> T.pack (show (e ^. tags)))) let url = "https://app.wallabag.it/api/entries/" <> show (e ^. identifier) <> "/tags" - _ <- liftIO $ postWith opts url ["tags" := T.intercalate "," (e ^. tags)] + metrics <- mkTracerMetrics + _ <- liftIO $ postWith metrics SafeTransient opts url ["tags" := T.intercalate "," (e ^. tags)] pure () diff --git a/src/Wallabag/Auth.hs b/src/Wallabag/Auth.hs new file mode 100644 index 0000000..35eaed4 --- /dev/null +++ b/src/Wallabag/Auth.hs @@ -0,0 +1,16 @@ +module Wallabag.Auth where + +import Control.Concurrent.STM (TVar) +import Data.Text (Text) +import Data.Time (UTCTime) + +data Auth' + = Unauthenticated + | Authenticating + | Auth + { accessToken :: Text, + expires :: UTCTime + } + deriving (Show) + +type Auth = TVar Auth' diff --git a/wallabag-tagger.cabal b/wallabag-tagger.cabal index df37472..826ecf2 100644 --- a/wallabag-tagger.cabal +++ b/wallabag-tagger.cabal @@ -22,16 +22,24 @@ library Creds M Wallabag + Wallabag.Auth MyLib + HTTP + Tracer -- other-modules: -- other-extensions: build-depends: base ^>=4.18.2.1 , wreq , lens , katip + , http-client + , http-types + , ekg-core + , retry , pipes , streamly , streamly-core + , unordered-containers , monad-control , transformers-base , pandoc