@@ -22,21 +22,34 @@ import (
2222
2323const (
2424 scheduleActionRecordTable = "scheduler.schedule_action_record"
25+ actionIdCol = "action_id"
2526 jobNameCol = "job_name"
2627 actionCol = "action"
2728 scheduledAtCol = "scheduled_at"
2829)
2930
3031// AddScheduleActionRecord adds a new schedule action record to the database
31- // Note: the created field should be set manually before calling this function, and all the records belong to the same job should have the same created time.
32- // So that the created time can be used to query the latest schedule action records of a job.
32+ // Note: the scheduledAt field should be set manually before calling this function.
3333func (c * Client ) AddScheduleActionRecord (ctx context.Context , scheduleActionRecord model.ScheduleActionRecord ) (model.ScheduleActionRecord , errors.EdgeX ) {
3434 if len (scheduleActionRecord .Id ) == 0 {
3535 scheduleActionRecord .Id = uuid .New ().String ()
3636 }
3737 return addScheduleActionRecord (ctx , c .ConnPool , scheduleActionRecord )
3838}
3939
40+ // AddScheduleActionRecords adds multiple schedule action records to the database
41+ func (c * Client ) AddScheduleActionRecords (ctx context.Context , scheduleActionRecords []model.ScheduleActionRecord ) ([]model.ScheduleActionRecord , errors.EdgeX ) {
42+ records := make ([]model.ScheduleActionRecord , len (scheduleActionRecords ))
43+ for _ , record := range scheduleActionRecords {
44+ r , err := c .AddScheduleActionRecord (ctx , record )
45+ if err != nil {
46+ return nil , errors .NewCommonEdgeXWrapper (err )
47+ }
48+ records = append (records , r )
49+ }
50+ return records , nil
51+ }
52+
4053// AllScheduleActionRecords queries the schedule action records with the given range, offset, and limit
4154func (c * Client ) AllScheduleActionRecords (ctx context.Context , start , end int64 , offset , limit int ) ([]model.ScheduleActionRecord , errors.EdgeX ) {
4255 var err errors.EdgeX
@@ -47,42 +60,32 @@ func (c *Client) AllScheduleActionRecords(ctx context.Context, start, end int64,
4760
4861 records , err := queryScheduleActionRecords (ctx , c .ConnPool , sqlQueryAllWithPaginationAndTimeRange (scheduleActionRecordTable ), time .UnixMilli (start ), time .UnixMilli (end ), offset , limit )
4962 if err != nil {
50- return nil , errors .NewCommonEdgeX (errors .KindDatabaseError , "failed to query all schedule action records" , err )
63+ return nil , errors .NewCommonEdgeX (errors .Kind ( err ) , "failed to query all schedule action records" , err )
5164 }
5265
5366 return records , nil
5467}
5568
56- // LatestScheduleActionRecords queries the latest schedule action records of all schedule jobs with the given offset and limit
57- func (c * Client ) LatestScheduleActionRecords (ctx context.Context , offset , limit int ) ([]model.ScheduleActionRecord , errors.EdgeX ) {
58- // Get all the job names
59- jobNames , err := queryScheduleJobNames (ctx , c .ConnPool )
60- if err != nil {
61- return nil , errors .NewCommonEdgeXWrapper (err )
62- }
63-
69+ // LatestScheduleActionRecordsByJobName queries the latest schedule action records by job name
70+ func (c * Client ) LatestScheduleActionRecordsByJobName (ctx context.Context , jobName string ) ([]model.ScheduleActionRecord , errors.EdgeX ) {
6471 sqlQueryLatestScheduleActionRecords := `
65- SELECT id, job_name, action, status, scheduled_at, created
72+ SELECT id, action_id, job_name, action, status, scheduled_at, created
6673 FROM(
6774 SELECT *
6875 FROM (
6976 SELECT *,
70- RANK() OVER (PARTITION BY job_name ORDER BY created DESC) AS rnk
77+ RANK() OVER (PARTITION BY job_name, action_id ORDER BY created DESC) AS rnk
7178 FROM scheduler.schedule_action_record
72- WHERE job_name = ANY($1::text[])
79+ WHERE job_name = $1
7380 ) subquery
7481 WHERE rnk = 1
7582 )
76- ORDER BY job_name, created DESC
77- OFFSET $2
78- LIMIT $3;
83+ ORDER BY job_name, created DESC;
7984 `
8085
81- // Pass the offset and limit here
82- offset , limit = getValidOffsetAndLimit (offset , limit )
83- records , err := queryScheduleActionRecords (ctx , c .ConnPool , sqlQueryLatestScheduleActionRecords , jobNames , offset , limit )
86+ records , err := queryScheduleActionRecords (ctx , c .ConnPool , sqlQueryLatestScheduleActionRecords , jobName )
8487 if err != nil {
85- return nil , errors .NewCommonEdgeX (errors .KindDatabaseError , "failed to query latest schedule action records" , err )
88+ return nil , errors .NewCommonEdgeX (errors .Kind ( err ) , "failed to query latest schedule action records" , err )
8689 }
8790
8891 return records , nil
@@ -98,7 +101,7 @@ func (c *Client) ScheduleActionRecordsByStatus(ctx context.Context, status strin
98101
99102 records , err := queryScheduleActionRecords (ctx , c .ConnPool , sqlQueryAllByStatusWithPaginationAndTimeRange (scheduleActionRecordTable ), status , time .UnixMilli (start ), time .UnixMilli (end ), offset , limit )
100103 if err != nil {
101- return nil , errors .NewCommonEdgeX (errors .KindDatabaseError , fmt .Sprintf ("failed to query schedule action records by status %s" , status ), err )
104+ return nil , errors .NewCommonEdgeX (errors .Kind ( err ) , fmt .Sprintf ("failed to query schedule action records by status %s" , status ), err )
102105 }
103106
104107 return records , nil
@@ -114,7 +117,7 @@ func (c *Client) ScheduleActionRecordsByJobName(ctx context.Context, jobName str
114117
115118 records , err := queryScheduleActionRecords (ctx , c .ConnPool , sqlQueryAllByColWithPaginationAndTimeRange (scheduleActionRecordTable , jobNameCol ), jobName , time .UnixMilli (start ), time .UnixMilli (end ), offset , limit )
116119 if err != nil {
117- return nil , errors .NewCommonEdgeX (errors .KindDatabaseError , fmt .Sprintf ("failed to query schedule action records by job name %s" , jobName ), err )
120+ return nil , errors .NewCommonEdgeX (errors .Kind ( err ) , fmt .Sprintf ("failed to query schedule action records by job name %s" , jobName ), err )
118121 }
119122
120123 return records , nil
@@ -130,7 +133,7 @@ func (c *Client) ScheduleActionRecordsByJobNameAndStatus(ctx context.Context, jo
130133
131134 records , err := queryScheduleActionRecords (ctx , c .ConnPool , sqlQueryAllByColWithPaginationAndTimeRange (scheduleActionRecordTable , jobNameCol , statusCol ), jobName , status , time .UnixMilli (start ), time .UnixMilli (end ), offset , limit )
132135 if err != nil {
133- return nil , errors .NewCommonEdgeX (errors .KindDatabaseError , fmt .Sprintf ("failed to query schedule action records by job name %s and status %s" , jobName , status ), err )
136+ return nil , errors .NewCommonEdgeX (errors .Kind ( err ) , fmt .Sprintf ("failed to query schedule action records by job name %s and status %s" , jobName , status ), err )
134137 }
135138
136139 return records , nil
@@ -141,23 +144,6 @@ func (c *Client) ScheduleActionRecordTotalCount(ctx context.Context) (uint32, er
141144 return getTotalRowsCount (ctx , c .ConnPool , sqlQueryCount (scheduleActionRecordTable ))
142145}
143146
144- // LatestScheduleActionRecordTotalCount returns the total count of all the latest schedule action records
145- func (c * Client ) LatestScheduleActionRecordTotalCount (ctx context.Context ) (uint32 , errors.EdgeX ) {
146- sqlQueryLatestScheduleActionRecordCount := `
147- SELECT COUNT(*)
148- FROM (
149- SELECT *
150- FROM (
151- SELECT *,
152- RANK() OVER (PARTITION BY job_name ORDER BY created DESC) AS rnk
153- FROM scheduler.schedule_action_record
154- )
155- WHERE rnk = 1
156- )
157- `
158- return getTotalRowsCount (ctx , c .ConnPool , sqlQueryLatestScheduleActionRecordCount )
159- }
160-
161147// ScheduleActionRecordCountByStatus returns the total count of the schedule action records by status
162148func (c * Client ) ScheduleActionRecordCountByStatus (ctx context.Context , status string ) (uint32 , errors.EdgeX ) {
163149 return getTotalRowsCount (ctx , c .ConnPool , sqlQueryCountByCol (scheduleActionRecordTable , statusCol ), status )
@@ -190,13 +176,13 @@ func addScheduleActionRecord(ctx context.Context, connPool *pgxpool.Pool, schedu
190176
191177 _ , err = connPool .Exec (
192178 ctx ,
193- sqlInsert (scheduleActionRecordTable , idCol , jobNameCol , actionCol , statusCol , scheduledAtCol , createdCol ),
179+ sqlInsert (scheduleActionRecordTable , idCol , actionIdCol , jobNameCol , actionCol , statusCol , scheduledAtCol ),
194180 scheduleActionRecord .Id ,
181+ copiedScheduleAction .GetBaseScheduleAction ().Id ,
195182 scheduleActionRecord .JobName ,
196183 actionJSONBytes ,
197184 scheduleActionRecord .Status ,
198- time .UnixMilli (scheduleActionRecord .ScheduledAt ).UTC (),
199- time .UnixMilli (scheduleActionRecord .Created ).UTC ())
185+ time .UnixMilli (scheduleActionRecord .ScheduledAt ).UTC ())
200186 if err != nil {
201187 return scheduleActionRecord , pgClient .WrapDBError ("failed to insert schedule action record" , err )
202188 }
@@ -213,10 +199,11 @@ func queryScheduleActionRecords(ctx context.Context, connPool *pgxpool.Pool, sql
213199
214200 var scheduleActionRecords []model.ScheduleActionRecord
215201 for rows .Next () {
202+ var actionId string
216203 var record model.ScheduleActionRecord
217204 var created , scheduledAt time.Time
218205 var actionJSONBytes []byte
219- err := rows .Scan (& record .Id , & record .JobName , & actionJSONBytes , & record .Status , & scheduledAt , & created )
206+ err := rows .Scan (& record .Id , & actionId , & record .JobName , & actionJSONBytes , & record .Status , & scheduledAt , & created )
220207 if err != nil {
221208 return nil , pgClient .WrapDBError ("failed to scan schedule action record" , err )
222209 }
0 commit comments