2019-05-11 06:54:12 +02:00
package main
import (
2019-05-11 06:55:05 +02:00
"encoding/json"
2019-05-11 06:54:12 +02:00
"fmt"
2019-05-27 05:11:16 +02:00
"strconv"
2019-05-11 06:55:05 +02:00
"time"
2019-05-11 06:54:12 +02:00
)
2019-05-27 05:08:10 +02:00
func createJob ( jobTypeID int32 , priority int32 , userID64 int64 , schedule time . Time , payload [ ] byte ) ( int64 , error ) {
stmt , err := db . Prepare ( ` INSERT INTO obj ( obj_type_id , obj_sub_type_id )
VALUES ( ? , ? ) ; ` )
logOnError ( err , "createJob : prepare insert obj" )
if err != nil {
return 0 , err
}
defer stmt . Close ( )
res , err := stmt . Exec ( objTypeJob , jobTypeID )
s := fmt . Sprintf ( "createJob, insert obj(%d, %d)" , objTypeJob , jobTypeID )
logOnError ( err , s )
if err != nil {
return 0 , err
}
objId , err := res . LastInsertId ( )
logOnError ( err , "createJob : get last insert Id" )
if err != nil {
return 0 , err
}
2019-06-03 03:19:06 +02:00
stmt , err = db . Prepare ( ` INSERT INTO obj_job ( obj_id , priority , user_id , status , seq_nr , schedule , is_done , in_work , inserted , pulled , started , ended , payload )
VALUES ( ? , ? , ? , ? , NULL , ? , 0 , 0 , ? , NULL , NULL , NULL , ? ) ; ` )
2019-05-27 05:08:10 +02:00
logOnError ( err , "createJob : prepare insert obj_job" )
if err != nil {
return 0 , err
}
defer stmt . Close ( )
_ , err = stmt . Exec ( objId , priority , userID64 , objJobStatusNew , schedule , time . Now ( ) , payload )
logOnError ( err , "createJob : insert obj_job" )
if err != nil {
return 0 , err
}
return objId , nil
}
func setJobDone ( jobId int64 ) error {
stmt , err := db . Prepare ( ` UPDATE obj_job j SET j.is_done = 1, j.in_work = 0, j.ended = ? WHERE j.obj_id = ?; ` )
logOnError ( err , "setJobDone : prepare update obj_job" )
if err != nil {
return err
}
defer stmt . Close ( )
_ , err = stmt . Exec ( time . Now ( ) , jobId )
s := fmt . Sprintf ( "setJobDone, update obj_job(%d)" , jobId )
logOnError ( err , s )
if err != nil {
return err
}
return nil
}
func setJobStart ( jobId int64 ) error {
stmt , err := db . Prepare ( ` UPDATE obj_job j SET j.started = coalesce(j.started, ?) WHERE j.obj_id = ?; ` )
logOnError ( err , "setJobStart : prepare update obj_job" )
if err != nil {
return err
}
defer stmt . Close ( )
_ , err = stmt . Exec ( time . Now ( ) , jobId )
s := fmt . Sprintf ( "setJobStart, update obj_job(%d)" , jobId )
logOnError ( err , s )
if err != nil {
return err
}
return nil
}
2019-05-27 06:03:04 +02:00
func rescheduleJob ( jobID64 int64 , status int32 , schedule time . Time ) error {
stmt , err := db . Prepare ( ` UPDATE obj_job j SET j.is_done = 0, j.in_work = 0, j.schedule = ?, j.status = ? WHERE j.obj_id = ?; ` )
2019-05-27 05:08:10 +02:00
logOnError ( err , "rescheduleJob : prepare update obj_job" )
if err != nil {
return err
}
defer stmt . Close ( )
2019-06-04 00:18:26 +02:00
_ , err = stmt . Exec ( schedule , status , jobID64 )
2019-05-27 05:11:16 +02:00
s := fmt . Sprintf ( "rescheduleJob, update obj_job(%d)" , jobID64 )
2019-05-27 05:08:10 +02:00
logOnError ( err , s )
if err != nil {
return err
}
return nil
}
2019-05-31 14:24:27 +02:00
func loadCurrentJobs ( ) ( [ ] Job , error ) {
2019-05-27 05:08:10 +02:00
var (
objId int64
jobTypeId int32
2019-05-30 04:05:43 +02:00
userID64 int64
2019-05-27 05:08:10 +02:00
status int32
payload [ ] byte
jobs [ ] Job
)
2019-06-02 12:33:55 +02:00
t := time . Now ( )
2019-06-03 03:19:06 +02:00
r := RndInt64 ( )
2019-06-02 12:33:55 +02:00
2019-06-03 03:19:06 +02:00
_ , err := db . Exec ( "UPDATE obj_job j SET j.pulled = ?, j.in_work = 1, j.seq_nr = ? WHERE j.is_done = 0 AND j.in_work = 0 AND j.schedule <= ? ORDER BY j.priority ASC, j.obj_id ASC LIMIT ?;" , t , r , t , SQLJobSliceSize )
2019-05-27 05:08:10 +02:00
logOnError ( err , "loadCurrentJobs : update intial rows" )
2019-06-03 03:19:06 +02:00
stmt , err := db . Prepare ( "SELECT o.id, o.obj_sub_type_id, j.status, j.user_id, j.payload FROM obj_job j, obj o WHERE j.obj_id = o.id AND j.is_done = 0 AND j.in_work = 1 AND j.seq_nr = ? ORDER BY j.priority ASC, j.obj_id ASC;" )
2019-05-27 05:08:10 +02:00
logOnError ( err , "loadCurrentJobs : prepare select statement" )
2019-06-03 03:19:06 +02:00
rows , err := stmt . Query ( r )
2019-05-27 05:08:10 +02:00
// rows, err := stmt.Query(time.Now())
logOnError ( err , "loadCurrentJobs : query select statement" )
for rows . Next ( ) {
2019-06-02 12:45:23 +02:00
err = rows . Scan ( & objId , & jobTypeId , & status , & userID64 , & payload )
2019-05-27 05:08:10 +02:00
logOnError ( err , "loadCurrentJobs : scan query rows" )
job := Job {
ID64 : objId ,
JobTypeID : jobTypeId ,
Status : status ,
2019-05-30 04:05:43 +02:00
UserID64 : userID64 ,
2019-05-27 05:08:10 +02:00
Payload : payload ,
}
jobs = append ( jobs , job )
}
err = rows . Err ( )
logOnError ( err , "loadCurrentJobs : scan end rows" )
rows . Close ( )
err = stmt . Close ( )
logOnError ( err , "loadCurrentJobs : close select statement" )
return jobs , nil
}
2019-05-11 06:54:49 +02:00
func jobRescan ( j Job ) {
2019-05-11 07:06:40 +02:00
var r JobPayloadRescanMsg
2019-05-11 06:54:12 +02:00
err := setJobStart ( j . ID64 )
logOnError ( err , "jobRescan : setJobStart" )
2019-05-11 07:06:40 +02:00
2019-05-11 06:56:48 +02:00
err = json . Unmarshal ( j . Payload , & r )
2019-05-11 07:06:40 +02:00
logOnError ( err , "jobRescan : Unmarshal payload" )
2019-05-19 05:24:45 +02:00
start := time . Now ( )
2019-05-11 06:54:12 +02:00
ids := getSQLListID64 ( r . Query )
2019-05-11 07:06:40 +02:00
2019-05-11 06:54:12 +02:00
if len ( ids ) > 1 {
2019-05-19 05:20:01 +02:00
for _ , id := range ids {
SQLMsgIdentifyQueue <- id
2019-05-11 06:54:12 +02:00
}
2019-05-19 05:20:01 +02:00
p := JobPayloadSetDone {
JobID64 : j . ID64 ,
MsgID64 : r . MsgID64 ,
ChatID64 : r . ChatID64 ,
2019-05-19 05:24:45 +02:00
Text : fmt . Sprintf ( "%d messages processed in %s." , len ( ids ) , time . Since ( start ) ) ,
2019-05-11 06:54:12 +02:00
}
2019-05-19 05:20:01 +02:00
b , _ := json . Marshal ( p )
2019-05-26 15:46:54 +02:00
_ , err := createJob ( objSubTypeJobSetJobDone , objJobPriorityRescanAllMsg , j . UserID64 , time . Now ( ) , b )
2019-05-19 05:20:01 +02:00
logOnError ( err , "jobRescan : createJob(objSubTypeJobSetJobDone)" )
2019-05-11 06:54:12 +02:00
} else if len ( ids ) == 1 {
SQLMsgIdentifyQueue <- ids [ 0 ]
err = setJobDone ( j . ID64 )
2019-05-11 07:06:40 +02:00
logOnError ( err , "jobRescan : setJobDone(1)" )
2019-05-16 05:06:38 +02:00
if r . MsgID64 != 0 || r . ChatID64 != 0 {
m := TGCommand {
Type : commandReplyMsg ,
Text : "One message processed." ,
FromMsgID64 : r . MsgID64 ,
FromChatID64 : r . ChatID64 ,
}
TGCmdQueue <- m
2019-05-16 04:49:34 +02:00
}
2019-05-11 07:06:40 +02:00
} else {
err = setJobDone ( j . ID64 )
logOnError ( err , "jobRescan : setJobDone(0)" )
2019-05-16 05:06:38 +02:00
if r . MsgID64 != 0 || r . ChatID64 != 0 {
m := TGCommand {
Type : commandReplyMsg ,
2019-05-16 05:11:56 +02:00
Text : "No message processed." ,
2019-05-16 05:06:38 +02:00
FromMsgID64 : r . MsgID64 ,
FromChatID64 : r . ChatID64 ,
}
TGCmdQueue <- m
2019-05-16 04:49:34 +02:00
}
2019-05-11 06:54:12 +02:00
}
return
}
2019-05-11 06:54:49 +02:00
func jobSetDone ( j Job ) {
2019-05-11 07:06:40 +02:00
var r JobPayloadSetDone
2019-05-11 06:54:12 +02:00
err := setJobStart ( j . ID64 )
logOnError ( err , "jobSetDone : setJobStart" )
2019-05-11 07:06:40 +02:00
2019-05-11 06:56:48 +02:00
err = json . Unmarshal ( j . Payload , & r )
2019-05-11 07:06:40 +02:00
logOnError ( err , "jobSetDone : Unmarshal payload" )
2019-05-11 06:54:12 +02:00
err = setJobDone ( r . JobID64 )
logOnError ( err , "jobSetDone : setJobDone(child)" )
2019-05-11 07:06:40 +02:00
2019-05-11 06:54:12 +02:00
err = setJobDone ( j . ID64 )
logOnError ( err , "jobSetDone : setJobDone" )
2019-05-11 07:06:40 +02:00
2019-05-16 04:52:30 +02:00
m := TGCommand {
Type : commandReplyMsg ,
Text : r . Text ,
FromMsgID64 : r . MsgID64 ,
FromChatID64 : r . ChatID64 ,
}
TGCmdQueue <- m
2019-05-11 06:54:12 +02:00
return
}
2019-05-16 14:39:12 +02:00
func jobPillage ( j Job ) {
2019-05-27 05:08:10 +02:00
var r JobPayloadPillage
err := setJobStart ( j . ID64 )
logOnError ( err , "jobPillage : setJobStart" )
err = json . Unmarshal ( j . Payload , & r )
logOnError ( err , "jobPillage : Unmarshal payload" )
// check if we have a acknoledgment of go or a timeout within 3m30 of the PillageInc from the Job
ids := getSQLListID64 ( ` select ox . id
from obj ox
, obj_msg omx
, obj op
, obj_msg omp
, obj_job oj
2019-05-27 05:20:05 +02:00
where oj . obj_id = ` + strconv.FormatInt(j.ID64, 10) + `
2019-05-27 05:08:10 +02:00
and omx . user_id = oj . user_id
and omx . sender_user_id = ` + strconv.Itoa(userID64ChtWrsBot) + `
and omx . obj_id = ox . id
2019-05-27 05:47:13 +02:00
and ox . obj_sub_type_id in ( ` + strconv . Itoa ( objSubTypeMessagePillageGo ) +
` , ` + strconv . Itoa ( objSubTypeMessagePillageTimeout ) +
` , ` + strconv . Itoa ( objSubTypeMessagePillageLoss ) +
` , ` + strconv . Itoa ( objSubTypeMessagePillageWin ) + ` )
2019-05-27 05:20:05 +02:00
and op . id = ` + strconv.FormatInt(r.ObjID64, 10) + `
2019-05-27 05:08:10 +02:00
and omp . obj_id = op . id
2019-05-27 05:47:13 +02:00
and omx . date between omp . date and addtime ( omp . date , ' 0 0 : 3 : 30.000000 ' )
order by case ox . obj_sub_type_id when ` + strconv.Itoa(objSubTypeMessagePillageWin) + ` then 0
when ` + strconv.Itoa(objSubTypeMessagePillageLoss) + ` then 1
when ` + strconv.Itoa(objSubTypeMessagePillageTimeout) + ` then 2
when ` + strconv.Itoa(objSubTypeMessagePillageGo) + ` then 3
else 4 end asc
limit 1 ; ` )
2019-05-27 05:08:10 +02:00
if len ( ids ) > 1 { // issue there ?
s := TGCommand {
Type : commandSendMsg ,
Text : fmt . Sprintf ( "More than one outcome for pillage #%d" , r . ObjID64 ) ,
2019-05-27 05:11:16 +02:00
ToUserID64 : j . UserID64 ,
2019-05-27 05:08:10 +02:00
}
TGCmdQueue <- s
} else if len ( ids ) == 1 { // we've got a match, job is done whether we prevented the pillage or not
m , err := getMsg ( ids [ 0 ] )
logOnError ( err , "jobPillage : getMsg(objSubTypeMessagePillageGo, objSubTypeMessagePillageTimeout)" )
if err == nil {
if m . Date . Add ( 60 * time . Second ) . After ( time . Now ( ) ) {
2019-05-27 05:20:05 +02:00
msgTypeID64 , err := getObjSubTypeId ( ids [ 0 ] )
2019-05-27 05:08:10 +02:00
logOnError ( err , "jobPillage : getObjSubTypeId" )
if err == nil {
2019-05-27 05:20:05 +02:00
if msgTypeID64 == objSubTypeMessagePillageGo {
2019-05-27 05:08:10 +02:00
s := TGCommand {
Type : commandSendMsg ,
Text : fmt . Sprintf ( "We avoided a pillage (%s)" , m . Date . Format ( time . RFC3339 ) ) ,
2019-05-27 05:11:16 +02:00
ToUserID64 : j . UserID64 ,
2019-05-27 05:08:10 +02:00
}
TGCmdQueue <- s
2019-05-27 05:47:13 +02:00
} else if msgTypeID64 == objSubTypeMessagePillageWin {
s := TGCommand {
Type : commandSendMsg ,
Text : fmt . Sprintf ( "We avoided a pillage (%s))" , m . Date . Format ( time . RFC3339 ) ) ,
ToUserID64 : j . UserID64 ,
}
TGCmdQueue <- s
} else if msgTypeID64 == objSubTypeMessagePillageLoss {
s := TGCommand {
Type : commandSendMsg ,
Text : fmt . Sprintf ( "We got pillaged (%s)" , m . Date . Format ( time . RFC3339 ) ) ,
ToUserID64 : j . UserID64 ,
}
TGCmdQueue <- s
2019-05-27 05:20:05 +02:00
} else if msgTypeID64 == objSubTypeMessagePillageTimeout {
2019-05-27 05:08:10 +02:00
s := TGCommand {
Type : commandSendMsg ,
Text : fmt . Sprintf ( "We got pillaged (%s)" , m . Date . Format ( time . RFC3339 ) ) ,
2019-05-27 05:11:16 +02:00
ToUserID64 : j . UserID64 ,
2019-05-27 05:08:10 +02:00
}
TGCmdQueue <- s
} else {
s := TGCommand {
Type : commandSendMsg ,
Text : fmt . Sprintf ( "We don't know what happened (%s)" , m . Date . Format ( time . RFC3339 ) ) ,
2019-05-27 05:11:16 +02:00
ToUserID64 : j . UserID64 ,
2019-05-27 05:08:10 +02:00
}
TGCmdQueue <- s
}
}
}
}
2019-05-27 05:20:05 +02:00
err = setJobDone ( j . ID64 )
2019-05-27 05:08:10 +02:00
logOnError ( err , "jobSetDone : setJobDone" )
return
}
2019-05-31 14:24:27 +02:00
s := TGCommand {
Type : commandSendMsg ,
2019-05-31 14:30:08 +02:00
Text : fmt . Sprintf ( "No outcome for the pillage yet" ) ,
2019-05-31 14:24:27 +02:00
ToUserID64 : j . UserID64 ,
}
TGCmdQueue <- s
2019-05-27 05:08:10 +02:00
//no outcome yet, have we sent a "/go" in the last 30 sec ?
ids = getSQLListID64 ( ` select ox . id
from obj ox
, obj_msg omx
, obj_job oj
2019-05-27 05:20:05 +02:00
where oj . obj_id = ` + strconv.FormatInt(j.ID64, 10) + `
2019-05-27 05:08:10 +02:00
and omx . user_id = oj . user_id
and omx . sender_user_id = oj . user_id
and omx . obj_id = ox . id
and ox . obj_sub_type_id = ` + strconv.Itoa(objSubTypeMessageGo) + `
and omx . date between addtime ( oj . schedule , ' - 30 ' ) and oj . schedule ; ` )
if len ( ids ) > 0 { // we did, so we reschedule the job to check the outcome and wait
m , err := getMsg ( ids [ 0 ] )
logOnError ( err , "jobPillage : getMsg(objSubTypeMessageGo)" )
if err == nil {
s := TGCommand {
Type : commandSendMsg ,
Text : fmt . Sprintf ( "We started intercepting the pillage (%s)" , m . Date . Format ( time . RFC3339 ) ) ,
2019-05-27 05:11:16 +02:00
ToUserID64 : j . UserID64 ,
2019-05-27 05:08:10 +02:00
}
TGCmdQueue <- s
}
2019-05-27 06:03:04 +02:00
err = rescheduleJob ( j . ID64 , j . Status + 1 , time . Now ( ) . Add ( 30 * time . Second ) )
2019-05-27 05:08:10 +02:00
logOnError ( err , "jobPillage : rescheduleJob(objSubTypeMessageGo)" )
} else { //no /go in the last 30 sec so we go ahead, send one and reschedule to check again in 25sec
2019-05-30 06:12:01 +02:00
clientSendCWMsg ( j . UserID64 , "/go" )
2019-05-27 06:03:04 +02:00
err = rescheduleJob ( j . ID64 , j . Status + 1 , time . Now ( ) . Add ( 30 * time . Second ) )
2019-05-27 05:08:10 +02:00
logOnError ( err , "jobPillage : rescheduleJob" )
}
2019-05-16 14:39:12 +02:00
return
}
2019-05-26 15:06:12 +02:00
func jobMsgClient ( j Job ) {
2019-05-30 06:12:01 +02:00
var p JobPayloadMsgClient
2019-05-26 15:06:12 +02:00
err := setJobStart ( j . ID64 )
logOnError ( err , "jobMsgClient : setJobStart" )
2019-05-30 06:12:01 +02:00
err = json . Unmarshal ( j . Payload , & p )
2019-05-26 15:06:12 +02:00
logOnError ( err , "jobMsgClient : Unmarshal payload" )
2019-05-31 03:16:40 +02:00
if err == nil {
2019-06-03 03:01:18 +02:00
clientSendCWMsg ( j . UserID64 , p . Text )
m := TGCommand {
Type : commandReplyMsg ,
Text : "Message sent." ,
FromMsgID64 : p . MsgID64 ,
FromChatID64 : p . ChatID64 ,
}
TGCmdQueue <- m
2019-05-30 06:12:01 +02:00
}
2019-05-26 15:06:12 +02:00
2019-05-26 15:07:28 +02:00
err = setJobDone ( j . ID64 )
2019-05-26 15:06:12 +02:00
logOnError ( err , "joMsgClient : setJobDone" )
return
}