bot.go

package main

import (
  	"log"
	"fmt"
	"time"

	"net/url"

  	"encoding/json"

	_ "github.com/mattn/go-sqlite3"
	"github.com/jmoiron/sqlx"
	"github.com/go-telegram-bot-api/telegram-bot-api"

	"gopkg.in/resty.v1"
)

var selectObservers = `SELECT userid FROM observers WHERE branch = ? and last_event < ?`
var insertObserver = `INSERT INTO observers(userid,branch,last_event) values(?,?,?)`
var deleteObserver = `DELETE FROM observers WHERE userid = ? and branch = ?`

type TGMsg struct {
	msg    string
	chatID int64
	branch string
	number int
}

type JenkinsJobs struct {
  Class json.RawMessage `json:"_class"`
  Jobs  []struct {
    Class     json.RawMessage `json:"_class"`
    Name      string `json:"name"`
    LastBuild json.RawMessage `json:"lastBuild"`
    LastStableBuild json.RawMessage `json:"lastStableBuild"`
  } `json:"jobs"`
}

type Build struct {
  Class       json.RawMessage `json:"_class"`
  Building    bool   `json:"building"`
  DisplayName string `json:"displayName"`
  Result      string `json:"result"`
  Timestamp   int64  `json:"timestamp"`
  Number   	  int    `json:"number"`
  // ChangeSets []json.RawMessage `json:"changeSets"`
}

type ChangeSets struct {
	Class      string `json:"_class"`
	ChangeSets []struct {
		Class string `json:"_class"`
		Items []struct {
			Class string `json:"_class"`
			Msg   string `json:"msg"`
		} `json:"items"`
	} `json:"changeSets"`
}

func initdb(db *sqlx.DB) {
	sqlStmt := `
		CREATE TABLE IF NOT EXISTS observers
		(
			id               INTEGER PRIMARY KEY AUTOINCREMENT,
			userid           TEXT,
			branch           TEXT,
			last_event       INTEGER, 
			UNIQUE(userid, branch) ON CONFLICT REPLACE
		);
	`
	db.MustExec(sqlStmt)
}

func watch(db *sqlx.DB, userid int64, branch string) string {
	tx := db.MustBegin()
	tx.MustExec(insertObserver, 
		userid, 
		branch, 
		0)
	tx.Commit()
	return "watch"
}  

func unwatch(db *sqlx.DB, userid int64, branch string) string {
	tx := db.MustBegin()
	tx.MustExec(deleteObserver, 
		userid, 
		branch)
	tx.Commit()
 	return "unwatch"
}

func doSomething(db *sqlx.DB, jobs *JenkinsJobs, tgMsgChan chan<- TGMsg) {
	
	for _, job := range jobs.Jobs {
		jobName, err := url.QueryUnescape(job.Name)
		if err == nil { 
			var stableBuild Build
			if err := json.Unmarshal(job.LastStableBuild, &stableBuild); err == nil {
				var ids []int64
				rows, _ := db.Query(selectObservers, jobName, stableBuild.Timestamp / 1000)
			    for rows.Next() {
			        var userid int64
			        err = rows.Scan(&userid)
			        if err == nil {
			        	ids = append(ids, userid)
			    	}
			    }
			    rows.Close()
				for i := 0; i < len(ids); i++ {
					db.MustExec(insertObserver, 
						ids[i], 
						jobName, 
						stableBuild.Timestamp / 1000)
					const emoji = "\xf0\x9f\x8e\x82"
					tgMsgChan <- TGMsg{fmt.Sprintf(`Новая сборка %s "%s" %s [%s]`, emoji, jobName, stableBuild.DisplayName, time.Unix(stableBuild.Timestamp / 1000,0).Format("02.01 15:04")), ids[i], jobName, stableBuild.Number}
				}
			 //    for elem := range users {
				// 	db.MustExec(insertObserver, 
				// 		elem, 
				// 		jobName, 
				// 		stableBuild.Timestamp)
				// }
			}
		} 
	}
}

func TGPolling(db *sqlx.DB, bot *tgbotapi.BotAPI, tgMsgChan chan<- TGMsg) {
	bot.Debug = false
	u := tgbotapi.NewUpdate(0)
	u.Timeout = 30
	updates, _ := bot.GetUpdatesChan(u)
	for update := range updates {
		if update.Message == nil {
			continue
		}
		// -------------------------------------
		if update.Message.IsCommand() {
			if (update.Message.From.ID != XXXXXXXXX) {
				tgMsgChan <- TGMsg{fmt.Sprintf("err your id %d", update.Message.Chat.ID), update.Message.Chat.ID, "", 0}
			} else {
				switch update.Message.Command() {
				case "help":
					tgMsgChan <- TGMsg{"type /watch branch_name", update.Message.Chat.ID, "", 0}
				// case "watch":
				// 	tgMsgChan <- TGMsg{watch(db, update.Message.Chat.ID, update.Message.CommandArguments()), update.Message.Chat.ID, "", 0}
				// case "unwatch":
				// 	tgMsgChan <- TGMsg{unwatch(db, update.Message.Chat.ID, update.Message.CommandArguments()), update.Message.Chat.ID, "", 0}
				// case "list":
				// 	tgMsgChan <- TGMsg{"list", update.Message.Chat.ID, "", 0}
				default:
					tgMsgChan <- TGMsg{fmt.Sprintf("I don't know that command %d", update.Message.Chat.ID), update.Message.Chat.ID, "", 0}
				}
			}
		}
	}
}

func JJPolling(db *sqlx.DB, tgMsgChan chan<- TGMsg) {
	for {
		resp, err := resty.R().
			SetResult(JenkinsJobs{}).
			Get("https://XXX-XXXXXXX.XXXXX.XXXX.ru/job/XXXXX/job/XXX/api/json?tree=jobs[name,lastStableBuild[status,timestamp,result,displayName,building,timestamp,number]]")
		if err == nil {
			go doSomething(db, resp.Result().(*JenkinsJobs), tgMsgChan)
		}
		<-time.After(1 * time.Minute)
	}
}

func addChanges(msg TGMsg, tgMsgChan chan<- string) {
	jurl := fmt.Sprintf("https://XXX-XXXXXXX.XXXXX.XXXX.XX/job/XXXXX/job/XXX/job/%s/%d/api/json?tree=changeSets[items[msg]]", url.QueryEscape(url.QueryEscape(msg.branch)), msg.number)
	resp, err := resty.R().
		SetResult(ChangeSets{}).
		Get(jurl)
	if err == nil {
		changeSets := resp.Result().(*ChangeSets)
		for j := range changeSets.ChangeSets {
			for i := range changeSets.ChangeSets[j].Items {
				msg.msg = msg.msg + "\n" + changeSets.ChangeSets[j].Items[i].Msg
			}
		}
		tgMsgChan <- msg.msg
	}
	<-time.After(30 * time.Second)
}

func main() {
	tgMsgChan := make(chan TGMsg)

	resty.SetBasicAuth("XXXXXXXXXXXXXX", "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX")
	resty.SetRESTMode()
	resty.SetTimeout(time.Duration(1 * time.Minute))

	bot, err := tgbotapi.NewBotAPI("XXXXXXXXX:XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX")
	if err != nil {
		log.Panic(err)
	}

	db := sqlx.MustOpen("sqlite3", "./jenkins_bot.db")
	if err != nil {
		log.Fatal(err)
	}
	db.SetMaxIdleConns(100)
	defer db.Close()

	initdb(db)

	go TGPolling(db, bot, tgMsgChan)
	go JJPolling(db, tgMsgChan)

	for elem := range tgMsgChan {
		<-time.After(1 * time.Second)
		msg := tgbotapi.NewMessage(elem.chatID, elem.msg)
		msgSended, err := bot.Send(msg)
		if elem.branch != "" && elem.number != 0 {
			if err == nil {
				tgEdtMsg := make(chan string, 1)
				go addChanges(elem, tgEdtMsg)
				edtMsg := tgbotapi.NewEditMessageText(msgSended.Chat.ID, msgSended.MessageID, <-tgEdtMsg)
				bot.Send(edtMsg)
			} else {
				db.MustExec(insertObserver, 
							elem.chatID, 
							elem.branch, 
							0)
			}
		} 
	}
}