module IHP.Job.Queue where
import IHP.Prelude
import IHP.Job.Types
import qualified Database.PostgreSQL.Simple as PG
import qualified Database.PostgreSQL.Simple.Types as PG
import qualified Database.PostgreSQL.Simple.FromField as PG
import qualified Database.PostgreSQL.Simple.ToField as PG
import qualified Database.PostgreSQL.Simple.Notification as PG
import qualified Control.Concurrent.Async as Async
import qualified Control.Concurrent as Concurrent
import IHP.ModelSupport
import IHP.QueryBuilder
import IHP.Fetch
import IHP.Controller.Param
import qualified System.Random as Random
import qualified IHP.PGListener as PGListener
fetchNextJob :: forall job.
( ?modelContext :: ModelContext
, job ~ GetModelByTableName (GetTableName job)
, FilterPrimaryKey (GetTableName job)
, FromRow job
, Show (PrimaryKey (GetTableName job))
, PG.FromField (PrimaryKey (GetTableName job))
, Table job
) => UUID -> IO (Maybe job)
fetchNextJob :: forall job.
(?modelContext::ModelContext,
job ~ GetModelByTableName (GetTableName job),
FilterPrimaryKey (GetTableName job), FromRow job,
Show (PrimaryKey (GetTableName job)),
FromField (PrimaryKey (GetTableName job)), Table job) =>
UUID -> IO (Maybe job)
fetchNextJob UUID
workerId = do
let query :: Query
query = Query
"UPDATE ? SET status = ?, locked_at = NOW(), locked_by = ?, attempts_count = attempts_count + 1 WHERE id IN (SELECT id FROM ? WHERE ((status = ?) OR (status = ? AND updated_at < NOW() + interval '30 seconds')) AND locked_by IS NULL AND run_at <= NOW() ORDER BY created_at LIMIT 1 FOR UPDATE) RETURNING id"
let params :: (Identifier, JobStatus, UUID, Identifier, JobStatus, JobStatus)
params = (Text -> Identifier
PG.Identifier (forall record. Table record => Text
tableName @job), JobStatus
JobStatusRunning, UUID
workerId, Text -> Identifier
PG.Identifier (forall record. Table record => Text
tableName @job), JobStatus
JobStatusNotStarted, JobStatus
JobStatusRetry)
[Only (Id' (GetTableName job))]
result :: [PG.Only (Id job)] <- forall q r.
(?modelContext::ModelContext, ToRow q, FromRow r) =>
Query -> q -> IO [r]
sqlQuery Query
query (Identifier, JobStatus, UUID, Identifier, JobStatus, JobStatus)
params
case [Only (Id' (GetTableName job))]
result of
[] -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a. Maybe a
Nothing
[PG.Only Id' (GetTableName job)
id] -> forall a. a -> Maybe a
Just forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> forall fetchable model.
(Fetchable fetchable model, Table model, FromRow model,
?modelContext::ModelContext) =>
fetchable -> IO (FetchResult fetchable model)
fetch Id' (GetTableName job)
id
[Only (Id' (GetTableName job))]
otherwise -> forall a. Text -> a
error (forall a. Show a => a -> Text
show [Only (Id' (GetTableName job))]
otherwise)
watchForJob :: (?modelContext :: ModelContext) => PGListener.PGListener -> Text -> Int -> Concurrent.MVar JobWorkerProcessMessage -> IO (PGListener.Subscription, Async.Async ())
watchForJob :: (?modelContext::ModelContext) =>
PGListener
-> Text
-> Int
-> MVar JobWorkerProcessMessage
-> IO (Subscription, Async ())
watchForJob PGListener
pgListener Text
tableName Int
pollInterval MVar JobWorkerProcessMessage
onNewJob = do
let tableNameBS :: ByteString
tableNameBS = forall a b. ConvertibleStrings a b => a -> b
cs Text
tableName
forall q.
(?modelContext::ModelContext, ToRow q) =>
Query -> q -> IO Int64
sqlExec (ByteString -> Query
createNotificationTrigger ByteString
tableNameBS) ()
Async ()
poller <- (?modelContext::ModelContext) =>
Text -> Int -> MVar JobWorkerProcessMessage -> IO (Async ())
pollForJob Text
tableName Int
pollInterval MVar JobWorkerProcessMessage
onNewJob
Subscription
subscription <- PGListener
pgListener forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> ByteString -> Callback -> PGListener -> IO Subscription
PGListener.subscribe (ByteString -> ByteString
channelName ByteString
tableNameBS) (forall a b. a -> b -> a
const (forall a. MVar a -> a -> IO ()
Concurrent.putMVar MVar JobWorkerProcessMessage
onNewJob JobWorkerProcessMessage
JobAvailable))
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Subscription
subscription, Async ()
poller)
pollForJob :: (?modelContext :: ModelContext) => Text -> Int -> Concurrent.MVar JobWorkerProcessMessage -> IO (Async.Async ())
pollForJob :: (?modelContext::ModelContext) =>
Text -> Int -> MVar JobWorkerProcessMessage -> IO (Async ())
pollForJob Text
tableName Int
pollInterval MVar JobWorkerProcessMessage
onNewJob = do
let query :: Query
query = Query
"SELECT COUNT(*) FROM ? WHERE ((status = ?) OR (status = ? AND updated_at < NOW() + interval '30 seconds')) AND locked_by IS NULL AND run_at <= NOW() LIMIT 1"
let params :: (Identifier, JobStatus, JobStatus)
params = (Text -> Identifier
PG.Identifier Text
tableName, JobStatus
JobStatusNotStarted, JobStatus
JobStatusRetry)
forall a. IO a -> IO (Async a)
Async.asyncBound do
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever do
Int
count :: Int <- forall q value.
(?modelContext::ModelContext, ToRow q, FromField value) =>
Query -> q -> IO value
sqlQueryScalar Query
query (Identifier, JobStatus, JobStatus)
params
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
count forall a. Ord a => a -> a -> Bool
> Int
0) (forall a. MVar a -> a -> IO ()
Concurrent.putMVar MVar JobWorkerProcessMessage
onNewJob JobWorkerProcessMessage
JobAvailable)
Int
jitter <- forall a (m :: * -> *). (Random a, MonadIO m) => (a, a) -> m a
Random.randomRIO (Int
0, Int
2000000)
let pollIntervalWithJitter :: Int
pollIntervalWithJitter = Int
pollInterval forall a. Num a => a -> a -> a
+ Int
jitter
Int -> IO ()
Concurrent.threadDelay Int
pollIntervalWithJitter
createNotificationTrigger :: ByteString -> PG.Query
createNotificationTrigger :: ByteString -> Query
createNotificationTrigger ByteString
tableName = ByteString -> Query
PG.Query forall a b. (a -> b) -> a -> b
$ ByteString
""
forall a. Semigroup a => a -> a -> a
<> ByteString
"BEGIN;\n"
forall a. Semigroup a => a -> a -> a
<> ByteString
"CREATE OR REPLACE FUNCTION " forall a. Semigroup a => a -> a -> a
<> ByteString
functionName forall a. Semigroup a => a -> a -> a
<> ByteString
"() RETURNS TRIGGER AS $$"
forall a. Semigroup a => a -> a -> a
<> ByteString
"BEGIN\n"
forall a. Semigroup a => a -> a -> a
<> ByteString
" PERFORM pg_notify('" forall a. Semigroup a => a -> a -> a
<> ByteString -> ByteString
channelName ByteString
tableName forall a. Semigroup a => a -> a -> a
<> ByteString
"', '');\n"
forall a. Semigroup a => a -> a -> a
<> ByteString
" RETURN new;"
forall a. Semigroup a => a -> a -> a
<> ByteString
"END;\n"
forall a. Semigroup a => a -> a -> a
<> ByteString
"$$ language plpgsql;"
forall a. Semigroup a => a -> a -> a
<> ByteString
"DROP TRIGGER IF EXISTS " forall a. Semigroup a => a -> a -> a
<> ByteString
insertTriggerName forall a. Semigroup a => a -> a -> a
<> ByteString
" ON " forall a. Semigroup a => a -> a -> a
<> ByteString
tableName forall a. Semigroup a => a -> a -> a
<> ByteString
"; CREATE TRIGGER " forall a. Semigroup a => a -> a -> a
<> ByteString
insertTriggerName forall a. Semigroup a => a -> a -> a
<> ByteString
" AFTER INSERT ON \"" forall a. Semigroup a => a -> a -> a
<> ByteString
tableName forall a. Semigroup a => a -> a -> a
<> ByteString
"\" FOR EACH ROW WHEN (NEW.status = 'job_status_not_started' OR NEW.status = 'job_status_retry') EXECUTE PROCEDURE " forall a. Semigroup a => a -> a -> a
<> ByteString
functionName forall a. Semigroup a => a -> a -> a
<> ByteString
"();\n"
forall a. Semigroup a => a -> a -> a
<> ByteString
"DROP TRIGGER IF EXISTS " forall a. Semigroup a => a -> a -> a
<> ByteString
updateTriggerName forall a. Semigroup a => a -> a -> a
<> ByteString
" ON " forall a. Semigroup a => a -> a -> a
<> ByteString
tableName forall a. Semigroup a => a -> a -> a
<> ByteString
"; CREATE TRIGGER " forall a. Semigroup a => a -> a -> a
<> ByteString
updateTriggerName forall a. Semigroup a => a -> a -> a
<> ByteString
" AFTER UPDATE ON \"" forall a. Semigroup a => a -> a -> a
<> ByteString
tableName forall a. Semigroup a => a -> a -> a
<> ByteString
"\" FOR EACH ROW WHEN (NEW.status = 'job_status_not_started' OR NEW.status = 'job_status_retry') EXECUTE PROCEDURE " forall a. Semigroup a => a -> a -> a
<> ByteString
functionName forall a. Semigroup a => a -> a -> a
<> ByteString
"();\n"
forall a. Semigroup a => a -> a -> a
<> ByteString
"COMMIT;"
where
functionName :: ByteString
functionName = ByteString
"notify_job_queued_" forall a. Semigroup a => a -> a -> a
<> ByteString
tableName
insertTriggerName :: ByteString
insertTriggerName = ByteString
"did_insert_job_" forall a. Semigroup a => a -> a -> a
<> ByteString
tableName
updateTriggerName :: ByteString
updateTriggerName = ByteString
"did_update_job_" forall a. Semigroup a => a -> a -> a
<> ByteString
tableName
channelName :: ByteString -> ByteString
channelName :: ByteString -> ByteString
channelName ByteString
tableName = ByteString
"job_available_" forall a. Semigroup a => a -> a -> a
<> ByteString
tableName
jobDidFail :: forall job.
( job ~ GetModelByTableName (GetTableName job)
, SetField "lockedBy" job (Maybe UUID)
, SetField "status" job JobStatus
, SetField "updatedAt" job UTCTime
, HasField "attemptsCount" job Int
, SetField "lastError" job (Maybe Text)
, Job job
, CanUpdate job
, Show job
, ?modelContext :: ModelContext
) => job -> SomeException -> IO ()
jobDidFail :: forall job.
(job ~ GetModelByTableName (GetTableName job),
SetField "lockedBy" job (Maybe UUID),
SetField "status" job JobStatus, SetField "updatedAt" job UTCTime,
HasField "attemptsCount" job Int,
SetField "lastError" job (Maybe Text), Job job, CanUpdate job,
Show job, ?modelContext::ModelContext) =>
job -> SomeException -> IO ()
jobDidFail job
job SomeException
exception = do
UTCTime
updatedAt <- IO UTCTime
getCurrentTime
Text -> IO ()
putStrLn (Text
"Failed job with exception: " forall a. Semigroup a => a -> a -> a
<> forall a. Show a => a -> Text
tshow SomeException
exception)
let ?job = job
job
let canRetry :: Bool
canRetry = forall model (name :: Symbol) value.
(KnownSymbol name, HasField name model value) =>
Proxy name -> model -> value
get forall a. IsLabel "attemptsCount" a => a
#attemptsCount job
job forall a. Ord a => a -> a -> Bool
< forall job. (Job job, ?job::job) => Int
maxAttempts
let status :: JobStatus
status = if Bool
canRetry then JobStatus
JobStatusRetry else JobStatus
JobStatusFailed
job
job
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall model (name :: Symbol) value.
(KnownSymbol name, SetField name model value) =>
Proxy name -> value -> model -> model
set forall a. IsLabel "status" a => a
#status JobStatus
status
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall model (name :: Symbol) value.
(KnownSymbol name, SetField name model value) =>
Proxy name -> value -> model -> model
set forall a. IsLabel "lockedBy" a => a
#lockedBy forall a. Maybe a
Nothing
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall model (name :: Symbol) value.
(KnownSymbol name, SetField name model value) =>
Proxy name -> value -> model -> model
set forall a. IsLabel "updatedAt" a => a
#updatedAt UTCTime
updatedAt
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall model (name :: Symbol) value.
(KnownSymbol name, SetField name model value) =>
Proxy name -> value -> model -> model
set forall a. IsLabel "lastError" a => a
#lastError (forall a. a -> Maybe a
Just (forall a. Show a => a -> Text
tshow SomeException
exception))
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall a. (CanUpdate a, ?modelContext::ModelContext) => a -> IO a
updateRecord
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
jobDidTimeout :: forall job.
( job ~ GetModelByTableName (GetTableName job)
, SetField "lockedBy" job (Maybe UUID)
, SetField "status" job JobStatus
, SetField "updatedAt" job UTCTime
, HasField "attemptsCount" job Int
, SetField "lastError" job (Maybe Text)
, Job job
, CanUpdate job
, Show job
, ?modelContext :: ModelContext
) => job -> IO ()
jobDidTimeout :: forall job.
(job ~ GetModelByTableName (GetTableName job),
SetField "lockedBy" job (Maybe UUID),
SetField "status" job JobStatus, SetField "updatedAt" job UTCTime,
HasField "attemptsCount" job Int,
SetField "lastError" job (Maybe Text), Job job, CanUpdate job,
Show job, ?modelContext::ModelContext) =>
job -> IO ()
jobDidTimeout job
job = do
UTCTime
updatedAt <- IO UTCTime
getCurrentTime
Text -> IO ()
putStrLn Text
"Job timed out"
let ?job = job
job
let canRetry :: Bool
canRetry = forall model (name :: Symbol) value.
(KnownSymbol name, HasField name model value) =>
Proxy name -> model -> value
get forall a. IsLabel "attemptsCount" a => a
#attemptsCount job
job forall a. Ord a => a -> a -> Bool
< forall job. (Job job, ?job::job) => Int
maxAttempts
let status :: JobStatus
status = if Bool
canRetry then JobStatus
JobStatusRetry else JobStatus
JobStatusTimedOut
job
job
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall model (name :: Symbol) value.
(KnownSymbol name, SetField name model value) =>
Proxy name -> value -> model -> model
set forall a. IsLabel "status" a => a
#status JobStatus
status
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall model (name :: Symbol) value.
(KnownSymbol name, SetField name model value) =>
Proxy name -> value -> model -> model
set forall a. IsLabel "lockedBy" a => a
#lockedBy forall a. Maybe a
Nothing
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall model (name :: Symbol) value.
(KnownSymbol name, SetField name model value) =>
Proxy name -> value -> model -> model
set forall a. IsLabel "updatedAt" a => a
#updatedAt UTCTime
updatedAt
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall model (name :: Symbol) value.
(KnownSymbol name, SetField name model (Maybe value)) =>
Proxy name -> value -> model -> model
setJust forall a. IsLabel "lastError" a => a
#lastError Text
"Timeout reached"
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall a. (CanUpdate a, ?modelContext::ModelContext) => a -> IO a
updateRecord
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
jobDidSucceed :: forall job.
( job ~ GetModelByTableName (GetTableName job)
, SetField "lockedBy" job (Maybe UUID)
, SetField "status" job JobStatus
, SetField "updatedAt" job UTCTime
, HasField "attemptsCount" job Int
, SetField "lastError" job (Maybe Text)
, Job job
, CanUpdate job
, Show job
, ?modelContext :: ModelContext
) => job -> IO ()
jobDidSucceed :: forall job.
(job ~ GetModelByTableName (GetTableName job),
SetField "lockedBy" job (Maybe UUID),
SetField "status" job JobStatus, SetField "updatedAt" job UTCTime,
HasField "attemptsCount" job Int,
SetField "lastError" job (Maybe Text), Job job, CanUpdate job,
Show job, ?modelContext::ModelContext) =>
job -> IO ()
jobDidSucceed job
job = do
Text -> IO ()
putStrLn Text
"Succeeded job"
UTCTime
updatedAt <- IO UTCTime
getCurrentTime
job
job
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall model (name :: Symbol) value.
(KnownSymbol name, SetField name model value) =>
Proxy name -> value -> model -> model
set forall a. IsLabel "status" a => a
#status JobStatus
JobStatusSucceeded
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall model (name :: Symbol) value.
(KnownSymbol name, SetField name model value) =>
Proxy name -> value -> model -> model
set forall a. IsLabel "lockedBy" a => a
#lockedBy forall a. Maybe a
Nothing
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall model (name :: Symbol) value.
(KnownSymbol name, SetField name model value) =>
Proxy name -> value -> model -> model
set forall a. IsLabel "updatedAt" a => a
#updatedAt UTCTime
updatedAt
forall {t1} {t2}. t1 -> (t1 -> t2) -> t2
|> forall a. (CanUpdate a, ?modelContext::ModelContext) => a -> IO a
updateRecord
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
instance PG.FromField JobStatus where
fromField :: FieldParser JobStatus
fromField Field
field (Just ByteString
"job_status_not_started") = forall (f :: * -> *) a. Applicative f => a -> f a
pure JobStatus
JobStatusNotStarted
fromField Field
field (Just ByteString
"job_status_running") = forall (f :: * -> *) a. Applicative f => a -> f a
pure JobStatus
JobStatusRunning
fromField Field
field (Just ByteString
"job_status_failed") = forall (f :: * -> *) a. Applicative f => a -> f a
pure JobStatus
JobStatusFailed
fromField Field
field (Just ByteString
"job_status_timed_out") = forall (f :: * -> *) a. Applicative f => a -> f a
pure JobStatus
JobStatusTimedOut
fromField Field
field (Just ByteString
"job_status_succeeded") = forall (f :: * -> *) a. Applicative f => a -> f a
pure JobStatus
JobStatusSucceeded
fromField Field
field (Just ByteString
"job_status_retry") = forall (f :: * -> *) a. Applicative f => a -> f a
pure JobStatus
JobStatusRetry
fromField Field
field (Just ByteString
value) = forall a err.
(Typeable a, Exception err) =>
(String -> Maybe Oid -> String -> String -> String -> err)
-> Field -> String -> Conversion a
PG.returnError String -> Maybe Oid -> String -> String -> String -> ResultError
PG.ConversionFailed Field
field (String
"Unexpected value for enum value. Got: " forall a. Semigroup a => a -> a -> a
<> forall a b. ConvertibleStrings a b => a -> b
cs ByteString
value)
fromField Field
field Maybe ByteString
Nothing = forall a err.
(Typeable a, Exception err) =>
(String -> Maybe Oid -> String -> String -> String -> err)
-> Field -> String -> Conversion a
PG.returnError String -> Maybe Oid -> String -> String -> String -> ResultError
PG.UnexpectedNull Field
field String
"Unexpected null for enum value"
instance Default JobStatus where
def :: JobStatus
def = JobStatus
JobStatusNotStarted
instance PG.ToField JobStatus where
toField :: JobStatus -> Action
toField JobStatus
JobStatusNotStarted = forall a. ToField a => a -> Action
PG.toField (Text
"job_status_not_started" :: Text)
toField JobStatus
JobStatusRunning = forall a. ToField a => a -> Action
PG.toField (Text
"job_status_running" :: Text)
toField JobStatus
JobStatusFailed = forall a. ToField a => a -> Action
PG.toField (Text
"job_status_failed" :: Text)
toField JobStatus
JobStatusTimedOut = forall a. ToField a => a -> Action
PG.toField (Text
"job_status_timed_out" :: Text)
toField JobStatus
JobStatusSucceeded = forall a. ToField a => a -> Action
PG.toField (Text
"job_status_succeeded" :: Text)
toField JobStatus
JobStatusRetry = forall a. ToField a => a -> Action
PG.toField (Text
"job_status_retry" :: Text)
instance InputValue JobStatus where
inputValue :: JobStatus -> Text
inputValue JobStatus
JobStatusNotStarted = Text
"job_status_not_started" :: Text
inputValue JobStatus
JobStatusRunning = Text
"job_status_running" :: Text
inputValue JobStatus
JobStatusFailed = Text
"job_status_failed" :: Text
inputValue JobStatus
JobStatusTimedOut = Text
"job_status_timed_out" :: Text
inputValue JobStatus
JobStatusSucceeded = Text
"job_status_succeeded" :: Text
inputValue JobStatus
JobStatusRetry = Text
"job_status_retry" :: Text
instance IHP.Controller.Param.ParamReader JobStatus where
readParameter :: ByteString -> Either ByteString JobStatus
readParameter = forall parameter.
(Enum parameter, InputValue parameter) =>
ByteString -> Either ByteString parameter
IHP.Controller.Param.enumParamReader