6208bb2 · odd-jobs · git.iko.soy

git.iko.soy/odd-jobs

Support qualified table names (i.e. [schame_name].[table_name]) (#53)

* fix createJobTable failure when table name contains dot `createJobTable` fails when the table name contains a dot for example `myschema.jobs`. * update TableName comment * use QualifiedIdentifier * Modified docs for TableName Co-authored-by: Saurabh Nanda <saurabh@vacationlabs.com>

commit6208bb201c883d75935d3dd089e8ab4795d9d64b authorKanagaraj M <kanagaraj@minio.io> dateSat 31 Oct 2020, 10:58 (6 years ago) parent75a0884

5 files changed+96−52

compare with master

unifiedsplit

src/OddJobs/ConfigBuilder.hs+1−1 src/OddJobs/Job.hs+36−27 src/OddJobs/Migrations.hs+40−18 src/OddJobs/Types.hs+17−4 src/OddJobs/Web.hs+2−2

src/OddJobs/ConfigBuilder.hs+1−1

@@ -188,7 +188,7 @@ defaultDynamicJobTypes :: TableName

188188 -> PGS.Query

189189 -> AllJobTypes

190190defaultDynamicJobTypes tname jobTypeSql = AJTSql $ \conn -> do

191− fmap (DL.map ((fromMaybe "(unknown)") . fromOnly)) $ PGS.query_ conn $ "select distinct(" <> jobTypeSql <> ") from " <> tname <> " order by 1 nulls last"

191+ fmap (DL.map ((fromMaybe "(unknown)") . fromOnly)) $ PGS.query conn ("select distinct(" <> jobTypeSql <> ") from ? order by 1 nulls last") (Only tname)

192192

193193-- | This makes __two important assumptions__. First, this /assumes/ that jobs

194194-- in your app are represented by a sum-type. For example:

src/OddJobs/Job.hs+36−27

@@ -245,8 +245,8 @@ concatJobDbColumns = concatJobDbColumns_ jobDbColumns ""

245245 concatJobDbColumns_ (col:cols) x = concatJobDbColumns_ cols (x <> col <> ", ")

246246

247247

248−findJobByIdQuery :: TableName -> PGS.Query

249−findJobByIdQuery tname = "SELECT " <> concatJobDbColumns <> " FROM " <> tname <> " WHERE id = ?"

248+findJobByIdQuery :: PGS.Query

249+findJobByIdQuery = "SELECT " <> concatJobDbColumns <> " FROM ? WHERE id = ?"

250250

251251withDbConnection :: (HasJobRunner m)

252252 => (Connection -> m a)

@@ -275,17 +275,17 @@ findJobById jid = do

275275 withDbConnection $ \conn -> liftIO $ findJobByIdIO conn tname jid

276276

277277findJobByIdIO :: Connection -> TableName -> JobId -> IO (Maybe Job)

278−findJobByIdIO conn tname jid = PGS.query conn (findJobByIdQuery tname) (Only jid) >>= \case

278+findJobByIdIO conn tname jid = PGS.query conn findJobByIdQuery (tname, jid) >>= \case

279279 [] -> pure Nothing

280280 [j] -> pure (Just j)

281281 js -> Prelude.error $ "Not expecting to find multiple jobs by id=" <> (show jid)

282282

283283

284−saveJobQuery :: TableName -> PGS.Query

285−saveJobQuery tname = "UPDATE " <> tname <> " set run_at = ?, status = ?, payload = ?, last_error = ?, attempts = ?, locked_at = ?, locked_by = ? WHERE id = ? RETURNING " <> concatJobDbColumns

284+saveJobQuery :: PGS.Query

285+saveJobQuery = "UPDATE ? set run_at = ?, status = ?, payload = ?, last_error = ?, attempts = ?, locked_at = ?, locked_by = ? WHERE id = ? RETURNING " <> concatJobDbColumns

286286

287−deleteJobQuery :: TableName -> PGS.Query

288−deleteJobQuery tname = "DELETE FROM " <> tname <> " WHERE id = ?"

287+deleteJobQuery :: PGS.Query

288+deleteJobQuery = "DELETE FROM ? WHERE id = ?"

289289

290290saveJob :: (HasJobRunner m) => Job -> m Job

291291saveJob j = do

@@ -294,8 +294,9 @@ saveJob j = do

294294

295295saveJobIO :: Connection -> TableName -> Job -> IO Job

296296saveJobIO conn tname Job{jobRunAt, jobStatus, jobPayload, jobLastError, jobAttempts, jobLockedBy, jobLockedAt, jobId} = do

297− rs <- PGS.query conn (saveJobQuery tname)

298− ( jobRunAt

297+ rs <- PGS.query conn saveJobQuery

298+ ( tname

299+ , jobRunAt

299300 , jobStatus

300301 , jobPayload

301302 , jobLastError

@@ -316,7 +317,7 @@ deleteJob jid = do

316317

317318deleteJobIO :: Connection -> TableName -> JobId -> IO ()

318319deleteJobIO conn tname jid = do

319− void $ PGS.execute conn (deleteJobQuery tname) (Only jid)

320+ void $ PGS.execute conn deleteJobQuery (tname, jid)

320321

321322runJobNowIO :: Connection -> TableName -> JobId -> IO (Maybe Job)

322323runJobNowIO conn tname jid = do

@@ -327,9 +328,9 @@ runJobNowIO conn tname jid = do

327328-- and somehow send an uninterruptibleCancel to that thread.

328329unlockJobIO :: Connection -> TableName -> JobId -> IO (Maybe Job)

329330unlockJobIO conn tname jid = do

330− fmap listToMaybe $ PGS.query conn q (Retry, jid, In [Locked])

331+ fmap listToMaybe $ PGS.query conn q (tname, Retry, jid, In [Locked])

331332 where

332− q = "update " <> tname <> " set status=?, run_at=now(), locked_at=null, locked_by=null where id=? and status in ? returning " <> concatJobDbColumns

333+ q = "update ? set status=?, run_at=now(), locked_at=null, locked_by=null where id=? and status in ? returning " <> concatJobDbColumns

333334

334335cancelJobIO :: Connection -> TableName -> JobId -> IO (Maybe Job)

335336cancelJobIO conn tname jid =

@@ -340,9 +341,9 @@ updateJobHelper :: TableName

340341 -> (Status, [Status], Maybe UTCTime, JobId)

341342 -> IO (Maybe Job)

342343updateJobHelper tname conn (newStatus, existingStates, mRunAt, jid) =

343− fmap listToMaybe $ PGS.query conn q (newStatus, runAt, jid, PGS.In existingStates)

344+ fmap listToMaybe $ PGS.query conn q (tname, newStatus, runAt, jid, PGS.In existingStates)

344345 where

345− q = "update " <> tname <> " set attempts=0, status=?, run_at=? where id=? and status in ? returning " <> concatJobDbColumns

346+ q = "update ? set attempts=0, status=?, run_at=? where id=? and status in ? returning " <> concatJobDbColumns

346347 runAt = case mRunAt of

347348 Nothing -> PGS.toField $ PGS.Identifier "run_at"

348349 Just t -> PGS.toField t

@@ -453,8 +454,8 @@ jobMonitor = do

453454 liftIO $ Dir.removePathForcibly f

454455

455456-- | Ref: 'jobPoller'

456−jobPollingSql :: TableName -> Query

457−jobPollingSql tname = "update " <> tname <> " set status = ?, locked_at = ?, locked_by = ?, attempts=attempts+1 WHERE id in (select id from " <> tname <> " where (run_at<=? AND ((status in ?) OR (status = ? and locked_at<?))) ORDER BY run_at ASC LIMIT 1 FOR UPDATE) RETURNING id"

457+jobPollingSql :: Query

458+jobPollingSql = "update ? set status = ?, locked_at = ?, locked_by = ?, attempts=attempts+1 WHERE id in (select id from ? where (run_at<=? AND ((status in ?) OR (status = ? and locked_at<?))) ORDER BY run_at ASC LIMIT 1 FOR UPDATE) RETURNING id"

458459

459460waitForJobs :: (HasJobRunner m)

460461 => m ()

@@ -508,8 +509,16 @@ jobPoller = do

508509 log LevelDebug $ LogText $ toS $ "[" <> processName <> "] Polling the job queue.."

509510 t <- liftIO getCurrentTime

510511 r <- liftIO $

511− PGS.query pollerDbConn (jobPollingSql tname)

512− (Locked, t, processName, t, (In [Queued, Retry]), Locked, (addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t))

512+ PGS.query pollerDbConn jobPollingSql

513+ ( tname

514+ , Locked

515+ , t

516+ , processName

517+ , tname

518+ , t

519+ , (In [Queued, Retry])

520+ , Locked

521+ , (addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t))

513522 case r of

514523 -- When we don't have any jobs to run, we can relax a bit...

515524 [] -> pure delayAction

@@ -537,8 +546,8 @@ jobEventListener = do

537546 concurrencyControlFn <- getConcurrencyControlFn

538547

539548 let tryLockingJob jid = do

540− let q = "UPDATE " <> tname <> " SET status=?, locked_at=now(), locked_by=?, attempts=attempts+1 WHERE id=? AND status in ? RETURNING id"

541− (withDbConnection $ \conn -> (liftIO $ PGS.query conn q (Locked, jwName, jid, In [Queued, Retry]))) >>= \case

549+ let q = "UPDATE ? SET status=?, locked_at=now(), locked_by=?, attempts=attempts+1 WHERE id=? AND status in ? RETURNING id"

550+ (withDbConnection $ \conn -> (liftIO $ PGS.query conn q (tname, Locked, jwName, jid, In [Queued, Retry]))) >>= \case

542551 [] -> do

543552 log LevelDebug $ LogText $ toS $ "Job was locked by someone else before I could start. Skipping it. JobId=" <> show jid

544553 pure Nothing

@@ -546,7 +555,7 @@ jobEventListener = do

546555 x -> error $ "WTF just happned? Was expecting a single row to be returned, received " ++ (show x)

547556

548557 withResource pool $ \monitorDbConn -> do

549− void $ liftIO $ PGS.execute monitorDbConn ("LISTEN " <> pgEventName tname) ()

558+ void $ liftIO $ PGS.execute monitorDbConn ("LISTEN ?") (Only $ pgEventName tname)

550559 forever $ do

551560 log LevelDebug $ LogText "[LISTEN/NOFIFY] Event loop"

552561 notif <- liftIO $ getNotification monitorDbConn

@@ -583,8 +592,8 @@ jobEventListener = do

583592

584593

585594

586−createJobQuery :: TableName -> PGS.Query

587−createJobQuery tname = "INSERT INTO " <> tname <> "(run_at, status, payload, last_error, attempts, locked_at, locked_by) VALUES (?, ?, ?, ?, ?, ?, ?) RETURNING " <> concatJobDbColumns

595+createJobQuery :: PGS.Query

596+createJobQuery = "INSERT INTO ? (run_at, status, payload, last_error, attempts, locked_at, locked_by) VALUES (?, ?, ?, ?, ?, ?, ?) RETURNING " <> concatJobDbColumns

588597

589598-- $createJobs

590599--

@@ -623,9 +632,9 @@ scheduleJob :: ToJSON p

623632 -> UTCTime -- ^ when should the job be executed

624633 -> IO Job

625634scheduleJob conn tname payload runAt = do

626− let args = ( runAt, Queued, toJSON payload, Nothing :: Maybe Value, 0 :: Int, Nothing :: Maybe Text, Nothing :: Maybe Text )

627− queryFormatter = toS <$> (PGS.formatQuery conn (createJobQuery tname) args)

628− rs <- PGS.query conn (createJobQuery tname) args

635+ let args = ( tname, runAt, Queued, toJSON payload, Nothing :: Maybe Value, 0 :: Int, Nothing :: Maybe Text, Nothing :: Maybe Text )

636+ queryFormatter = toS <$> (PGS.formatQuery conn createJobQuery args)

637+ rs <- PGS.query conn createJobQuery args

629638 case rs of

630639 [] -> (Prelude.error . (<> "Not expecting a blank result set when creating a job. Query=")) <$> queryFormatter

631640 [r] -> pure r

@@ -684,5 +693,5 @@ fetchAllJobRunners :: (MonadIO m)

684693 => Config

685694 -> m [JobRunnerName]

686695fetchAllJobRunners Config{cfgTableName, cfgDbPool} = liftIO $ withResource cfgDbPool $ \conn -> do

687− fmap (mapMaybe fromOnly) $ PGS.query_ conn $ "select distinct locked_by from " <> cfgTableName

696+ fmap (mapMaybe fromOnly) $ PGS.query conn "select distinct locked_by from ?" (Only cfgTableName)

688697

src/OddJobs/Migrations.hs+40−18

@@ -5,11 +5,12 @@ module OddJobs.Migrations

55where

77import Database.PostgreSQL.Simple as PGS

8+import Database.PostgreSQL.Simple.Types as PGS

89import Data.Functor (void)

910import OddJobs.Types

1011

11−createJobTableQuery :: TableName -> Query

12−createJobTableQuery tname = "CREATE TABLE " <> tname <>

12+createJobTableQuery :: Query

13+createJobTableQuery = "CREATE TABLE ?" <>

1314 "( id serial primary key" <>

1415 ", created_at timestamp with time zone default now() not null" <>

1516 ", updated_at timestamp with time zone default now() not null" <>

@@ -22,28 +23,49 @@ createJobTableQuery tname = "CREATE TABLE " <> tname <>

2223 ", locked_by text null" <>

2324 ", constraint incorrect_locking_info CHECK ((status <> 'locked' and locked_at is null and locked_by is null) or (status = 'locked' and locked_at is not null and locked_by is not null))" <>

2425 ");" <>

25− "create index idx_" <> tname <> "_created_at on " <> tname <> "(created_at);" <>

26− "create index idx_" <> tname <> "_updated_at on " <> tname <> "(updated_at);" <>

27− "create index idx_" <> tname <> "_locked_at on " <> tname <> "(locked_at);" <>

28− "create index idx_" <> tname <> "_locked_by on " <> tname <> "(locked_by);" <>

29− "create index idx_" <> tname <> "_status on " <> tname <> "(status);" <>

30− "create index idx_" <> tname <> "_run_at on " <> tname <> "(run_at);"

26+ "create index ? on ?(created_at);" <>

27+ "create index ? on ?(updated_at);" <>

28+ "create index ? on ?(locked_at);" <>

29+ "create index ? on ?(locked_by);" <>

30+ "create index ? on ?(status);" <>

31+ "create index ? on ?(run_at);"

3132

32−createNotificationTrigger :: TableName -> Query

33−createNotificationTrigger tname = "create or replace function " <> fnName <> "() returns trigger as $$" <>

33+createNotificationTrigger :: Query

34+createNotificationTrigger = "create or replace function ?() returns trigger as $$" <>

3435 "begin \n" <>

35− " perform pg_notify('" <> pgEventName tname <> "', \n" <>

36+ " perform pg_notify('?', \n" <>

3637 " json_build_object('id', new.id, 'run_at', new.run_at, 'locked_at', new.locked_at)::text); \n" <>

3738 " return new; \n" <>

3839 "end; \n" <>

3940 "$$ language plpgsql;" <>

40− "create trigger " <> trgName <> " after insert on " <> tname <> " for each row execute procedure " <> fnName <> "();"

41− where

42− fnName = "notify_job_monitor_for_" <> tname

43− trgName = "trg_notify_job_monitor_for_" <> tname

44−

41+ "create trigger ? after insert on ? for each row execute procedure ?();"

4542

4643createJobTable :: Connection -> TableName -> IO ()

4744createJobTable conn tname = void $ do

48− PGS.execute_ conn (createJobTableQuery tname)

49− PGS.execute_ conn (createNotificationTrigger tname)

45+ let tnameTxt = getTnameTxt tname

46+ PGS.execute conn createJobTableQuery

47+ ( tname

48+ , PGS.Identifier $ "idx_" <> tnameTxt <> "_created_at"

49+ , tname

50+ , PGS.Identifier $ "idx_" <> tnameTxt <> "_updated_at"

51+ , tname

52+ , PGS.Identifier $ "idx_" <> tnameTxt <> "_locked_at"

53+ , tname

54+ , PGS.Identifier $ "idx_" <> tnameTxt <> "_locked_by"

55+ , tname

56+ , PGS.Identifier $ "idx_" <> tnameTxt <> "_status"

57+ , tname

58+ , PGS.Identifier $ "idx_" <> tnameTxt <> "_run_at"

59+ , tname

60+ )

61+ PGS.execute conn createNotificationTrigger

62+ ( fnName

63+ , pgEventName tname

64+ , trgName

65+ , tname

66+ , fnName

67+ )

68+ where

69+ fnName = PGS.Identifier $ "notify_job_monitor_for_" <> (getTnameTxt tname)

70+ trgName = PGS.Identifier $ "trg_notify_job_monitor_for_" <> (getTnameTxt tname)

71+ getTnameTxt (PGS.QualifiedIdentifier _ tname') = tname'

src/OddJobs/Types.hs+17−4

@@ -7,6 +7,7 @@

77module OddJobs.Types where

99import Database.PostgreSQL.Simple as PGS

10+import Database.PostgreSQL.Simple.Types as PGS

1011import UnliftIO (MonadIO)

1112import UnliftIO.Concurrent (threadDelay)

1213import Data.Text.Conversions

@@ -23,7 +24,8 @@ import Lucid (Html(..))

2324import Data.Pool (Pool)

2425import Control.Monad.Logger (LogLevel)

2526

26−-- | An alias for 'Query' type. Since this type has an instance of 'IsString'

27+-- | An alias for 'QualifiedIdentifier' type. It is used for the job table name.

28+-- Since this type has an instance of 'IsString',

2729-- you do not need to do anything special to create a value for this type. Just

2830-- ensure you have the @OverloadedStrings@ extention enabled. For example:

2931--

@@ -33,10 +35,21 @@ import Control.Monad.Logger (LogLevel)

3335-- myJobsTable :: TableName

3436-- myJobsTable = "my_jobs"

3537-- @

36−type TableName = PGS.Query

38+--

39+-- This should also work for table names qualified by the schema name. For example:

40+--

41+-- @

42+-- {-\# LANGUAGE OverloadedStrings \#-}

43+--

44+-- myJobsTable :: TableName

45+-- myJobsTable = "odd_jobs.jobs"

46+-- @

47+

48+type TableName = PGS.QualifiedIdentifier

3749

38−pgEventName :: TableName -> Query

39−pgEventName tname = "job_created_" <> tname

50+pgEventName :: TableName -> PGS.Identifier

51+pgEventName (PGS.QualifiedIdentifier Nothing tname) = PGS.Identifier $ "jobs_created_" <> tname

52+pgEventName (PGS.QualifiedIdentifier (Just schema) tname) = PGS.Identifier $ "jobs_created_" <> schema <> "_" <> tname

4053

4154newtype Seconds = Seconds { unSeconds :: Int } deriving (Eq, Show, Ord, Num, Read)

4255

src/OddJobs/Web.hs+2−2

@@ -128,8 +128,8 @@ data Routes = Routes

128128

129129filterJobsQuery :: Config -> Filter -> (PGS.Query, [Action])

130130filterJobsQuery Config{cfgTableName, cfgJobTypeSql} Filter{..} =

131− ( "SELECT " <> Job.concatJobDbColumns <> " FROM " <> cfgTableName <> whereClause <> " " <> (orderClause $ fromMaybe (OrdUpdatedAt, Desc) filterOrder) <> " " <> limitOffsetClause

132− , whereActions

131+ ( "SELECT " <> Job.concatJobDbColumns <> " FROM ?" <> whereClause <> " " <> (orderClause $ fromMaybe (OrdUpdatedAt, Desc) filterOrder) <> " " <> limitOffsetClause

132+ , (toRow $ Only cfgTableName) ++ whereActions

133133 )

134134 where

135135 orderClause (flt, dir) =