260 lines
6.9 KiB
Go
260 lines
6.9 KiB
Go
package common
|
|
|
|
/*
|
|
* This file is part of Autogits.
|
|
*
|
|
* Copyright © 2024 SUSE LLC
|
|
*
|
|
* Autogits is free software: you can redistribute it and/or modify it under
|
|
* the terms of the GNU General Public License as published by the Free Software
|
|
* Foundation, either version 2 of the License, or (at your option) any later
|
|
* version.
|
|
*
|
|
* Autogits is distributed in the hope that it will be useful, but WITHOUT ANY
|
|
* WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
|
|
* PARTICULAR PURPOSE. See the GNU General Public License for more details.
|
|
*
|
|
* You should have received a copy of the GNU General Public License along with
|
|
* Foobar. If not, see <https://www.gnu.org/licenses/>.
|
|
*/
|
|
|
|
import (
|
|
"crypto/tls"
|
|
"fmt"
|
|
"log"
|
|
"net/url"
|
|
"runtime/debug"
|
|
"slices"
|
|
"strings"
|
|
"time"
|
|
|
|
rabbitmq "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
const RequestType_CreateBrachTag = "create"
|
|
const RequestType_DeleteBranchTag = "delete"
|
|
const RequestType_Fork = "fork"
|
|
const RequestType_Issue = "issues"
|
|
const RequestType_IssueAssign = "issue_assign"
|
|
const RequestType_IssueComment = "issue_comment"
|
|
const RequestType_IssueLabel = "issue_label"
|
|
const RequestType_IssueMilestone = "issue_milestone"
|
|
const RequestType_Push = "push"
|
|
const RequestType_Repository = "repository"
|
|
const RequestType_Release = "release"
|
|
const RequestType_PR = "pull_request"
|
|
const RequestType_PRAssign = "pull_request_assign"
|
|
const RequestType_PRLabel = "pull_request_label"
|
|
const RequestType_PRComment = "pull_request_comment"
|
|
const RequestType_PRMilestone = "pull_request_milestone"
|
|
const RequestType_PRSync = "pull_request_sync"
|
|
const RequestType_PRReviewAccepted = "pull_request_review_approved"
|
|
const RequestType_PRReviewRejected = "pull_request_review_rejected"
|
|
const RequestType_PRReviewRequest = "pull_request_review_request"
|
|
const RequestType_PRReviewComment = "pull_request_review_comment"
|
|
const RequestType_Wiki = "wiki"
|
|
|
|
type RequestProcessor interface {
|
|
ProcessFunc(*Request) error
|
|
}
|
|
|
|
type ListenDefinitions struct {
|
|
RabbitURL string // amqps://user:password@host/queue
|
|
|
|
GitAuthor string
|
|
Handlers map[string]RequestProcessor
|
|
}
|
|
|
|
type RabbitMessage rabbitmq.Delivery
|
|
|
|
func processRabbitMQ(msgCh chan<- RabbitMessage, server url.URL, topics []string) error {
|
|
queueName := server.Path
|
|
server.Path = ""
|
|
|
|
if len(queueName) > 0 && queueName[0] == '/' {
|
|
queueName = queueName[1:]
|
|
}
|
|
|
|
connection, err := rabbitmq.DialTLS(server.String(), &tls.Config{
|
|
ServerName: server.Hostname(),
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("Cannot connect to %s . Err: %w", server.Hostname(), err)
|
|
}
|
|
defer connection.Close()
|
|
|
|
ch, err := connection.Channel()
|
|
if err != nil {
|
|
return fmt.Errorf("Cannot create a channel. Err: %w", err)
|
|
}
|
|
defer ch.Close()
|
|
|
|
if err = ch.ExchangeDeclarePassive("pubsub", "topic", true, false, false, false, nil); err != nil {
|
|
return fmt.Errorf("Cannot find pubsub exchange? Err: %w", err)
|
|
}
|
|
|
|
var q rabbitmq.Queue
|
|
if len(queueName) == 0 {
|
|
q, err = ch.QueueDeclare("", false, true, true, false, nil)
|
|
} else {
|
|
q, err = ch.QueueDeclarePassive(queueName, true, false, true, false, nil)
|
|
if err != nil {
|
|
log.Printf("queue not found .. trying to create it: %v\n", err)
|
|
if ch.IsClosed() {
|
|
ch, err = connection.Channel()
|
|
if err != nil {
|
|
return fmt.Errorf("Channel cannot be re-opened. Err: %w", err)
|
|
}
|
|
}
|
|
q, err = ch.QueueDeclare(queueName, true, false, true, false, nil)
|
|
|
|
if err != nil {
|
|
log.Printf("can't create persistent queue ... falling back to temporaty queue: %v\n", err)
|
|
if ch.IsClosed() {
|
|
ch, err = connection.Channel()
|
|
return fmt.Errorf("Channel cannot be re-opened. Err: %w", err)
|
|
}
|
|
q, err = ch.QueueDeclare("", false, true, true, false, nil)
|
|
}
|
|
}
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("Cannot declare queue. Err: %w", err)
|
|
}
|
|
// log.Printf("queue: %s:%d", q.Name, q.Consumers)
|
|
|
|
log.Println(" -- listening to topics:")
|
|
for _, topic := range topics {
|
|
err = ch.QueueBind(q.Name, topic, "pubsub", false, nil)
|
|
log.Println(" +", topic)
|
|
if err != nil {
|
|
return fmt.Errorf("Cannot find queue to exchange with topic %s. Err: %w", topic, err)
|
|
}
|
|
}
|
|
|
|
msgs, err := ch.Consume(q.Name, "", true, true, false, false, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("Cannot start consumer. Err: %w", err)
|
|
}
|
|
// log.Printf("queue: %s:%d", q.Name, q.Consumers)
|
|
|
|
for {
|
|
msg, ok := <-msgs
|
|
if !ok {
|
|
return fmt.Errorf("channel/connection closed?\n")
|
|
}
|
|
|
|
msgCh <- RabbitMessage(msg)
|
|
}
|
|
}
|
|
|
|
func connectAndProcessRabbitMQ(log *log.Logger, ch chan<- RabbitMessage, server url.URL, topics []string) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Println(r)
|
|
log.Println("'crash' RabbitMQ worker. Recovering... reconnecting...")
|
|
time.Sleep(5 * time.Second)
|
|
go connectAndProcessRabbitMQ(log, ch, server, topics)
|
|
}
|
|
}()
|
|
|
|
for {
|
|
err := processRabbitMQ(ch, server, topics)
|
|
if err != nil {
|
|
log.Printf("Error in RabbitMQ connection. %#v", err)
|
|
log.Println("Reconnecting in 2 seconds...")
|
|
time.Sleep(2 * time.Second)
|
|
}
|
|
}
|
|
}
|
|
|
|
func connectToRabbitMQ(log *log.Logger, server url.URL, topics []string) chan RabbitMessage {
|
|
ch := make(chan RabbitMessage, 100)
|
|
go connectAndProcessRabbitMQ(log, ch, server, topics)
|
|
|
|
return ch
|
|
}
|
|
|
|
func ProcessEvent(f RequestProcessor, request *Request) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Println(r)
|
|
log.Println(string(debug.Stack()))
|
|
}
|
|
}()
|
|
|
|
if err := f.ProcessFunc(request); err != nil {
|
|
log.Println(err)
|
|
}
|
|
|
|
}
|
|
|
|
func ProcessRabbitMQEvents(listenDefs ListenDefinitions, orgs []string) error {
|
|
server, err := url.Parse(listenDefs.RabbitURL)
|
|
if err != nil {
|
|
log.Panicf("cannot parse server URL. Err: %#v\n", err)
|
|
}
|
|
|
|
log.Println("RabbitMQ connection:", *server)
|
|
topics := make([]string, 0, len(listenDefs.Handlers)*len(orgs))
|
|
log.Println(len(listenDefs.Handlers), len(orgs))
|
|
|
|
server.User = url.UserPassword(rabbitUser, rabbitPassword)
|
|
|
|
scope := "suse"
|
|
if server.Hostname() == "rabbit.opensuse.org" {
|
|
scope = "opensuse"
|
|
}
|
|
|
|
for _, org := range orgs {
|
|
for k := range listenDefs.Handlers {
|
|
topics = append(topics, fmt.Sprintf("%s.src.%s.%s.#", scope, org, k))
|
|
}
|
|
}
|
|
|
|
slices.Sort(topics)
|
|
topics = slices.Compact(topics)
|
|
|
|
ch := connectToRabbitMQ(log.Default(), *server, topics)
|
|
|
|
for {
|
|
msg, ok := <-ch
|
|
if !ok {
|
|
return nil
|
|
}
|
|
|
|
log.Println("event:", msg.RoutingKey)
|
|
|
|
route := strings.Split(msg.RoutingKey, ".")
|
|
if len(route) > 3 {
|
|
reqType := route[3]
|
|
org := route[2]
|
|
|
|
if !slices.Contains(orgs, org) {
|
|
log.Println("Got even for unhandeled org:", org)
|
|
continue
|
|
}
|
|
|
|
log.Println("org:", org, "type:", reqType)
|
|
if handler, found := listenDefs.Handlers[reqType]; found {
|
|
/* h, err := CreateRequestHandler()
|
|
if err != nil {
|
|
log.Println("Cannot create request handler", err)
|
|
continue
|
|
}
|
|
*/
|
|
req, err := ParseRequestJSON(reqType, msg.Body)
|
|
if err != nil {
|
|
log.Println("Error parsing request JSON:", err)
|
|
continue
|
|
} else {
|
|
log.Println("processing req", req.Type)
|
|
// h.Request = req
|
|
ProcessEvent(handler, req)
|
|
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|