git.iko.soy/sc-build
on master help sign in
sc-build/src/Control/Task/Scheduler/Query.hs
154 lines 4.3 kB history raw
1{-# LANGUAGE TypeApplications #-}
3module Control.Task.Scheduler.Query
4 ( scheduleTaskQuery,
5 completeTaskQuery,
6 rescheduleTaskQuery,
7 pickTasksQuery,
8 makeAllRunnable,
9 PickedTask (..),
10 )
11where
13import Colog
14import Control.Arrow
15import Control.Monad.IO.Unlift
16import Control.Squeal
17import Control.Task
18import Data.Aeson hiding (Result (..))
19import Data.Proxy
20import qualified Data.Text as T
21import Data.Time
22import Data.Word
23import GHC.Generics (Generic)
24import GHC.TypeLits
25import qualified Generics.SOP as SOP
26import Server.Schema
27import Squeal.PostgreSQL
29scheduleTaskQuery ::
30 forall m t s env.
31 (StaticPQ m, Task t s, WithLog env Message m, MonadUnliftIO m) =>
32 t ->
33 m ()
34scheduleTaskQuery task = do
35 logInfo . T.pack $ "schdeuling task: " <> show task
36 let query =
37 insertInto
38 #tasks
39 ( Values_ $
40 Set (param @1) `as` #task
41 :* Set (param @2) `as` #payload
42 :* Default `as` #creation_time
43 :* Default `as` #started
44 :* Default `as` #failures
45 )
46 ( OnConflict
47 (OnConstraint #pk_task_payload)
48 (DoUpdate (Default `as` #creation_time) [])
49 )
50 (Returning_ Nil)
51 dbWrite (const $ return ()) $
52 manipulateParams_
53 query
54 (symbolVal @s Proxy, Jsonb $ toJSON task)
56completeTaskQuery ::
57 forall m t s env.
58 (StaticPQ m, Task t s, WithLog env Message m, MonadUnliftIO m) =>
59 t ->
60 m ()
61completeTaskQuery payload = do
62 let task = symbolVal @s Proxy
63 logInfo . T.pack $ "Completing task: " <> task <> " with payload " <> show payload
64 let query :: Manipulation_ Schema (String, Jsonb Value) ()
65 query =
66 deleteFrom_
67 #tasks
68 (param @1 .== #task .&& param @2 .== #payload)
69 dbWrite (const $ return ()) $
70 manipulateParams_
71 query
72 (symbolVal @s Proxy, Jsonb $ toJSON payload)
74rescheduleTaskQuery ::
75 forall s t m env.
76 (StaticPQ m, Task t s, WithLog env Message m, MonadUnliftIO m) =>
77 t ->
78 m ()
79rescheduleTaskQuery payload = do
80 let task = symbolVal @s Proxy
81 logInfo . T.pack $
82 "reschdeuling task: " <> task
83 <> " with payload "
84 <> show payload
85 let query :: Manipulation_ Schema (String, Jsonb Value) ()
86 query =
87 update_
88 #tasks
89 ( Default `as` #creation_time
90 :* Default `as` #started
91 :* Set (#failures + 1) `as` #failures
92 )
93 (param @1 .== #task .&& param @2 .== #payload)
94 dbWrite (const $ return ()) $
95 manipulateParams_
96 query
97 (symbolVal @s Proxy, Jsonb $ toJSON payload)
99pickTasksQuery ::
100 forall m env.
101 (StaticPQ m, WithLog env Message m, MonadUnliftIO m) =>
102 [String] ->
103 Word64 ->
104 m [PickedTask]
105pickTasksQuery tasks limitCount = do
106 logDebug $ "reading tasks " <> T.unwords (T.pack <$> tasks)
107 let query :: Query_ Schema () PickedTask
108 query =
109 select_
110 ( #task `as` #task
111 :* #payload `as` #payload
112 :* #creation_time `as` #creationTime
113 )
114 ( from (table #tasks)
115 & where_ (#task `in_` (literal <$> tasks))
116 & where_ (not_ . notNull $ #started)
117 & where_ (#failures .< 20)
118 & orderBy [#creation_time & Asc]
119 & limit limitCount
120 & orderBy [#creation_time & Desc]
121 )
122 startedMan :: Manipulation_ Schema (String, Jsonb Value) ()
123 startedMan =
124 update_
125 #tasks
126 ( Set true `as` #started
127 )
128 (param @1 .== #task .&& param @2 .== #payload)
129 dbWrite (const $ return []) $ do
130 pickedTasks <- runQuery query >>= getRows
131 traversePrepared_ startedMan $ (task &&& payload) <$> pickedTasks
132 return pickedTasks
134data PickedTask
135 = PickedTask
136 { task :: String,
137 payload :: Jsonb Value,
138 creationTime :: UTCTime
139 }
140 deriving (Eq, Show, Generic, SOP.Generic, SOP.HasDatatypeInfo)
142makeAllRunnable ::
143 (StaticPQ m, WithLog env Message m, MonadUnliftIO m) =>
144 m ()
145makeAllRunnable = do
146 let startAll :: Manipulation_ Schema () ()
147 startAll =
148 update_
149 #tasks
150 ( Set false `as` #started
151 )
152 (notNull #started)
153 dbWrite (const $ return ()) $
154 manipulate_ startAll