diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,3 @@ +* 0.1.0.0 Daniel Patterson 2015-10-25 + + Initial release. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,13 @@ +Copyright (c) 2015 Daniel Patterson + +Permission to use, copy, modify, and/or distribute this software for any purpose +with or without fee is hereby granted, provided that the above copyright notice +and this permission notice appear in all copies. + +THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND +FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS +OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER +TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF +THIS SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,74 @@ +## About + +This is a library that facilitates sending email via AWS SES using the +background processor [hworker](http://hackage.haskell.org/package/hworker). In +particular, it handles rate limiting (for sending rate currently, not +daily quotas), as SES does not queue messages. + +## Rate limiting + +Aside from sending emails in the background, the main thing that +`hworker-ses` provides is rate limiting. The rate limiting is only on +a per-process (most likely, per server) basis, as it is controlled via +in-memory storage. This means that you should take into account how +many workers you are likely to have when setting the +messages-per-second rate on the workers. + +For example, if your rate limit is 30 messages per second, and you expect +to have at most 5 workers (you can organize them however you like, but a +simple strategy is just to start a worker with each application process), +then if you set the rate limit to be 6 messages per second, things will be +fine. + +Note that if you do run over, it won't mean that your messages actually +don't get delivered. That error code will trigger a retry on the message, +so eventually it will go out. But it's wasteful (as you can know in advance +if a given message will actually be able to be sent), and if you really +mess it up (ie, set an absurdly high limit), you potentially will have +your API calls rate limited, which isn't good. + +## Usage + +The program in the `example` directory is probably most of what you +need to get started. In particular, you create an `hworker` (you need +to give it a name for the queue - this should be shared by any servers +that should be accessing the same queue and sending the same messages, +though if you are sharing the redis server with other applications, +this should be distinct). The third argument is the per-second rate +limit for this server, and the last argument is the address that the +messages are sent from. + +Using what you get back, you can spawn workers and a monitor. Then you +can queue as many messages as you want, and they will be sent out by +the workers. + +You need to start at least one worker thread and at least +one monitor thread. You can have as many of these as you want, though +generally speaking, having more than one monitor thread per server is +probably overkill (for the system to work, you want at least one +monitor running _somewhere_ - so if a whole server crashes, you want +there to be a monitor elsewhere). Having many workers makes things run +faster. In particular, if you have a high sending limit, you will +probably want several workers. + +As specified in the [hworker](http://hackage.haskell.org/package/hworker) +documentation, the semantics of this queue is at-least-once, so it's +possible that messages can get sent multiple times in error conditions +(like if an entire server crashes right after the message is sent, but +before the fact that it was sent is acknowledged). But, provided that +redis is still available and is reliable, messages that have been +queued are guaranteed to get sent eventually (even in the case of +servers crashing, etc). The only caveat to that is that for the +message to get sent at least one worker and one monitor thread must be +running (thoses don't need to be running all the time, but as long as +they aren't running, messages might be delayed. Once they're started again, +those delayed messages will go out). + +## Primary Libraries Used + +- hworker +- amazonka-ses + +## Contributors + +- Daniel Patterson diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/hworker-ses.cabal b/hworker-ses.cabal new file mode 100644 --- /dev/null +++ b/hworker-ses.cabal @@ -0,0 +1,32 @@ +name: hworker-ses +version: 0.1.0.0 +synopsis: Library for sending email with Amazon's SES and hworker +description: See README. +homepage: http://github.com/dbp/hworker-ses +license: ISC +license-file: LICENSE +author: Daniel Patterson +maintainer: dbp@dbpmail.net +-- copyright: +category: Web +build-type: Simple +extra-source-files: README.md CHANGELOG.md +cabal-version: >=1.10 + +library + hs-source-dirs: src + exposed-modules: System.Hworker.SES + build-depends: base >= 4.7 && < 5 + , hworker + , time + , aeson + , text + , amazonka + , amazonka-ses + , lens + , unordered-containers + default-language: Haskell2010 + +source-repository head + type: git + location: https://github.com/dbp/hworker-ses diff --git a/src/System/Hworker/SES.hs b/src/System/Hworker/SES.hs new file mode 100644 --- /dev/null +++ b/src/System/Hworker/SES.hs @@ -0,0 +1,152 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} +module System.Hworker.SES ( SESWorker + , SESWorkerWith + , SESState + , SESJob(..) + , SESConfig(..) + , RedisConnection(..) + , defaultSESConfig + , create + , createWith + , queue + , worker + , monitor + , jobs + ) where + +import Control.Applicative ((<|>)) +import Control.Arrow ((&&&)) +import Control.Concurrent (threadDelay) +import Control.Concurrent.MVar (MVar, newMVar, putMVar, takeMVar) +import Control.Exception (SomeException, catch) +import Control.Lens (set) +import Control.Monad (mzero) +import Control.Monad.Trans.AWS hiding (page) +import Data.Aeson (FromJSON (..), ToJSON (..), + Value (Object, String), object, (.:), + (.=)) +import qualified Data.HashMap.Strict as HashMap +import Data.Monoid ((<>)) +import Data.Text (Text) +import qualified Data.Text as T +import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime) +import GHC.Generics +import Network.AWS.SES hiding (Success) +import System.Hworker hiding (create, createWith) +import qualified System.Hworker as Hworker (create, createWith) + +type SESWorkerWith a = Hworker (SESState a) (SESJob a) +type SESWorker = SESWorkerWith () + +data SESState a = SESState { sesLimit :: Int + , sesSource :: Text + , sesRecents :: MVar [UTCTime] + , sesAfter :: SESJob a -> IO () + , sesLogger :: forall a. Show a => a -> IO () + } + +data SESJob a = SESJob { sesEmTo :: Text + , sesEmSubj :: Text + , sesEmBodyText :: Maybe Text + , sesEmBodyHtml :: Maybe Text + , sesPayload :: Either Value a + } + deriving (Generic, Show) +instance ToJSON a => ToJSON (SESJob a) where + toJSON SESJob{..} = object [ "v" .= (1 :: Int) + , "t" .= sesEmTo + , "s" .= sesEmSubj + , "x" .= sesEmBodyText + , "h" .= sesEmBodyHtml + , ("p", either id toJSON sesPayload) + ] +instance FromJSON a => FromJSON (SESJob a) where + parseJSON (Object v) = SESJob <$> v .: "t" + <*> v .: "s" + <*> v .: "x" + <*> v .: "h" + <*> ((Right <$> (v .: "p")) <|> + pure (Left (HashMap.lookupDefault (String "No 'p' field.") "p" v))) + parseJSON _ = mzero + +instance (ToJSON a, FromJSON a, Show a) => Job (SESState a) (SESJob a) where + job state@(SESState limit source recents after log) j@(SESJob to subj btxt bhtml payload) = + do now <- getCurrentTime + rs <- takeMVar recents + let active = filter ((< 1) . diffUTCTime now) rs + let count = length active + if count >= limit + then putMVar recents active >> threadDelay 100000 >> job state j + else do putMVar recents (now : active) + awsenv <- getEnv NorthVirginia Discover + r <- runAWST awsenv $ + send (sendEmail source + (set dToAddresses [to] + destination) + (message (content subj) + (set bHtml + (content <$> bhtml) $ + set bText + (content <$> btxt) + body))) + case r of + Left err -> + do log err + return (Failure (T.pack (show err))) + Right _ -> do catch (after j) + (\(e::SomeException) -> + log ("hworker-ses callback raised exception: " <> show e)) + return Success + +data SESConfig a = + SESConfig { sesconfigName :: Text + , sesconfigLimit :: Int + , sesconfigSource :: Text + , sesconfigAfter :: (SESJob a -> IO ()) + , sesconfigLogger :: (forall a. Show a => a -> IO ()) + , sesconfigFailedQueueSize :: Int + , sesconfigRedisConnectInfo :: RedisConnection + } + +defaultSESConfig :: (ToJSON a, FromJSON a, Show a) + => Text + -> Int + -> Text + -> (SESJob a -> IO ()) + -> SESConfig a +defaultSESConfig name limit source after = + let d = defaultHworkerConfig name () + in SESConfig name + limit + source + after + (hwconfigLogger d) + (hwconfigFailedQueueSize d) + (hwconfigRedisConnectInfo d) + +create :: (ToJSON a, FromJSON a, Show a) + => Text + -> Int + -> Text + -> (SESJob a -> IO ()) + -> IO (SESWorkerWith a) +create name limit source after = + createWith (defaultSESConfig name limit source after) + +createWith :: (ToJSON a, FromJSON a, Show a) + => SESConfig a + -> IO (SESWorkerWith a) +createWith (SESConfig name limit source after logger failed redis) = + do recents <- newMVar [] + Hworker.createWith + (defaultHworkerConfig name + (SESState limit source recents after logger)) { + hwconfigRedisConnectInfo = redis + ,hwconfigFailedQueueSize = failed + ,hwconfigLogger = logger + }