test

def.go

@@ -245,6 +245,7 @@ const (
 	objSubTypeMessageUnionWar        = 349 // from Chat Wars Reports (not done)
 	objSubTypeMessageTUReportReq     = 350 // /tu_report (not done)
 	objSubTypeMessageTUReportAck     = 351 // result from /tu_report (not done)
+	objSubTypeMessageTimeout         = 352 // generic timeout for action
 	objSubTypeJobPillage             = 601
 	objSubTypeJobTribute             = 602
 	objSubTypeJobStatus              = 603

job.go

@@ -80,15 +80,15 @@ func setJobStart(jobId int64) error {
 	return nil
 }
 
-func rescheduleJob(jobID64 int64, schedule time.Time) error {
+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 = ? WHERE j.obj_id = ?;`)
+	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 = ?;`)
 	logOnError(err, "rescheduleJob : prepare update obj_job")
 	if err != nil {
 		return err
 	}
 	defer stmt.Close()
 
-	_, err = stmt.Exec(time.Now(), jobID64)
+	_, err = stmt.Exec(time.Now(), status, jobID64)
 	s := fmt.Sprintf("rescheduleJob, update obj_job(%d)", jobID64)
 	logOnError(err, s)
 	if err != nil {
@@ -335,7 +335,7 @@ func jobPillage(j Job) {
 			}
 			TGCmdQueue <- s
 		}
-		err = rescheduleJob(j.ID64, time.Now().Add(30*time.Second))
+		err = rescheduleJob(j.ID64, j.Status+1, time.Now().Add(30*time.Second))
 		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
 		c := TGCommand{
@@ -345,7 +345,7 @@ func jobPillage(j Job) {
 			ToChatID64:   userID64ChtWrsBot,
 		}
 		MQTGCmdQueue <- c
-		err = rescheduleJob(j.ID64, time.Now().Add(30*time.Second))
+		err = rescheduleJob(j.ID64, j.Status+1, time.Now().Add(30*time.Second))
 		logOnError(err, "jobPillage : rescheduleJob")
 
 	}

rules.go

@@ -118,6 +118,7 @@ func resetMsgParsingRules() {
 								,(5000, ` + strconv.Itoa(objSubTypeMessageUnionWar) + `, "Union war rep with both", "^(🏠Trade union news:\n)(?P<Union>(?s:.*))(🎪State of exclusive rights:\n)(?P<Fair>(?s:.*))$")
 								,(4900, ` + strconv.Itoa(objSubTypeMessageUnionWar) + `, "Union war rep only union", "^(🏠Trade union news:\n)(?P<Union>(?s:.*))(🎪State of exclusive rights:\n){0}(?P<Fair>(?s:.*))$")
 								,(4900, ` + strconv.Itoa(objSubTypeMessageUnionWar) + `, "Union war rep only fair", "^(🏠Trade union news:\n){0}(?P<Union>(?s:.*))(🎪State of exclusive rights:\n)(?P<Fair>(?s:.*))$")
+								,(5000, ` + strconv.Itoa(objSubTypeMessageTimeout) + `, "Generic timeout", "^Too late. Action is not available.$"
 		;`)
 	failOnError(err, "resetMsgParsingRules : populate table msg_rules")
 }

sql.go

@@ -365,6 +365,7 @@ func initDB() {
 						        ,(` + strconv.Itoa(objSubTypeMessageTUReportReq) + `, "tureport_req", "Trade Union War Report Req", ` + strconv.Itoa(objTypeMessage) + `)
 						        ,(` + strconv.Itoa(objSubTypeMessageTUReportAck) + `, "tureport_ack", "Trade Union War Report Ack", ` + strconv.Itoa(objTypeMessage) + `)
 						        ,(` + strconv.Itoa(objSubTypeMessageUnionWar) + `, "union_war", "Union war report", ` + strconv.Itoa(objTypeMessage) + `)
+						        ,(` + strconv.Itoa(objSubTypeMessageTimeout) + `, "timeout", "Generic timeout", ` + strconv.Itoa(objTypeMessage) + `)
 						        ,(` + strconv.Itoa(objSubTypeJobPillage) + `, "job_pillage", "Pillage job", ` + strconv.Itoa(objTypeJob) + `)
 						        ,(` + strconv.Itoa(objSubTypeJobTribute) + `, "job_tribute", "Tribute job", ` + strconv.Itoa(objTypeJob) + `)
 						        ,(` + strconv.Itoa(objSubTypeJobWithdrawal) + `, "job_withdraw", "Withdrawal job", ` + strconv.Itoa(objTypeJob) + `)

workers.go

@@ -110,13 +110,7 @@ func SQLIdentifyMsgWorker(id int, objIds <-chan int64) {
 					logOnError(err, "SQLIdentifyMsgWorker["+strconv.Itoa(id)+"] : insertMsgPillageInc")
 					err = setObjSubTypeId(objId, objSubTypeMessagePillageInc)
 					logOnError(err, "SQLIdentifyMsgWorker["+strconv.Itoa(id)+"] : setObjSubTypeId(PillageInc)")
-					p := JobPayloadPillage{
+					// only catch live pillages
-						ObjID64: objId,
-					}
-					b, _ := json.Marshal(&p)
-					_, err = createJob(objSubTypeJobPillage, objJobPriority, m.UserID64, m.Date.Add(time.Duration(25+rand.Intn(35))*time.Second), b)
-					logOnError(err, "SQLIdentifyMsgWorker["+strconv.Itoa(id)+"] : createJob(JobPillage)")
-					// only advertise live pillages
 					if m.Date.Add(3 * time.Minute).After(time.Now()) {
 						s := TGCommand{
 							Type:       commandSendMsg,
@@ -124,6 +118,12 @@ func SQLIdentifyMsgWorker(id int, objIds <-chan int64) {
 							ToUserID64: m.UserID64,
 						}
 						TGCmdQueue <- s
+						p := JobPayloadPillage{
+							ObjID64: objId,
+						}
+						b, _ := json.Marshal(&p)
+						_, err = createJob(objSubTypeJobPillage, objJobPriority, m.UserID64, m.Date.Add(time.Duration(25+rand.Intn(35))*time.Second), b)
+						logOnError(err, "SQLIdentifyMsgWorker["+strconv.Itoa(id)+"] : createJob(JobPillage)")
 					}
 				case objSubTypeMessageGo:
 					err = setObjSubTypeId(objId, objSubTypeMessageGo)
@@ -331,10 +331,10 @@ func MQTGCmdWorker(id int, cmds <-chan TGCommand) {
 			logOnError(err, "MQTGCmdWorker["+strconv.Itoa(id)+"] : Marshal(c)")
 			log.Printf("MQTGCmdWorker["+strconv.Itoa(id)+"] : new command.\n%s\n", string(j))
 			err = clientsQueue[c.FromUserID64].Channel.Publish(
-				"", // exchange
+				"",                                      // exchange
 				clientsQueue[c.FromUserID64].Queue.Name, // routing key
-				false, // mandatory
+				false,                                   // mandatory
-				false, // immediate
+				false,                                   // immediate
 				amqp.Publishing{
 					ContentType: "application/json",
 					Body:        []byte(j),