From a3b8c3cd3b0b6b946441a03e238d373c365edaef6845505845d4ec3c103e9ea7 Mon Sep 17 00:00:00 2001
From: Adam Majer <amajer@suse.com>
Date: Wed, 17 Jul 2024 17:20:24 +0200
Subject: [PATCH] .

---
 obs-staging-bot/main.go     |   6 ++-
 obs-staging-bot/rabbitmq.go | 101 ++++++++++++++++++++++++++++++++++++
 2 files changed, 105 insertions(+), 2 deletions(-)
 create mode 100644 obs-staging-bot/rabbitmq.go

diff --git a/obs-staging-bot/main.go b/obs-staging-bot/main.go
index 01ceb27..7c60433 100644
--- a/obs-staging-bot/main.go
+++ b/obs-staging-bot/main.go
@@ -4,7 +4,6 @@ import (
 	"time"
 
 	"src.opensuse.org/autogits/common"
-	"github.com/rabbitmq/amqp091-go"
 )
 
 const (
@@ -37,5 +36,8 @@ func main() {
 	common.RequireGiteaSecretToken()
 	common.RequireObsSecretToken()
 
-	
+	go startProcessingObsMessages("rabbit.opensuse.org", "opensuse", "opensuse")
+
+	stuck := make(chan int) 
+	<-stuck
 }
diff --git a/obs-staging-bot/rabbitmq.go b/obs-staging-bot/rabbitmq.go
new file mode 100644
index 0000000..c087960
--- /dev/null
+++ b/obs-staging-bot/rabbitmq.go
@@ -0,0 +1,101 @@
+package main
+
+import (
+	"crypto/tls"
+	"encoding/json"
+	"log"
+	"strings"
+	"time"
+
+	rabbitmq "github.com/rabbitmq/amqp091-go"
+)
+
+type BuildNotification struct {
+	BuildSuccess                                                          bool
+	Project, Package, Repository, Arch, Release, Rev, Buildtype, Workerid string
+	Starttime, Endtime, Readytime                                         uint64
+}
+
+func failOnError(err error, msg string) {
+	if err != nil {
+		log.Panicf("%s: %s", err, msg)
+	}
+}
+
+func processObsMessage(msg *rabbitmq.Delivery) {
+	key := strings.SplitN(msg.RoutingKey, ".", 4)
+	if len(key) != 4 || len(key[3]) < 7 || key[3][:6] != "build_" {
+		return
+	}
+
+	buildSuccess := false
+	switch key[3][6:] {
+	case "success", "unchanged":
+		buildSuccess = true
+	case "fail":
+		buildSuccess = false
+	default:
+		log.Printf("unknown build_ logging message: %s\n", msg.RoutingKey)
+		return
+	}
+
+	notification := &BuildNotification{
+		BuildSuccess: buildSuccess,
+	}
+	err := json.Unmarshal(msg.Body, notification)
+	if err != nil {
+		log.Printf("Cannot unmarshall json object: %s\n", msg.Body)
+		return
+	}
+
+	log.Printf("%v\n", notification)
+
+}
+
+func startProcessingObsMessages(host, username, password string) {
+	auth := ""
+	if len(username) > 0 && len(password) > 0 {
+		auth = username + ":" + password + "@"
+	}
+	connection, err := rabbitmq.DialTLS("amqps://"+auth+host, &tls.Config{
+		ServerName: host,
+	})
+	failOnError(err, "Cannot connect to rabbit.opensuse.org")
+	defer connection.Close()
+
+	ch, err := connection.Channel()
+	failOnError(err, "Cannot create a channel")
+	defer ch.Close()
+
+	err = ch.ExchangeDeclarePassive("pubsub", "topic", true, false, false, false, nil)
+	failOnError(err, "Cannot find pubsub exchange")
+
+	q, err := ch.QueueDeclare("", false, true, false, false, nil)
+	failOnError(err, "Cannot declare queue")
+	log.Printf("queue: %s:%d", q.Name, q.Consumers)
+
+	err = ch.QueueBind(q.Name, "*.obs.package.*", "pubsub", false, nil)
+	failOnError(err, "Cannot bind queue to exchange")
+
+	msgs, err := ch.Consume(q.Name, "", true, true, false, false, nil)
+	failOnError(err, "Cannot start consumer")
+	log.Printf("queue: %s:%d", q.Name, q.Consumers)
+
+	// TODO: add recovery for NotifyClose and related
+	for {
+		msg, ok := <-msgs
+		if !ok {
+			log.Printf("channel closed?\n")
+
+			if connection.IsClosed() {
+				// reconnect
+				log.Printf("reconnecting...")
+				time.Sleep(5 * time.Second)
+				go startProcessingObsMessages(host, username, password)
+			}
+			return
+		}
+
+		processObsMessage(&msg)
+	}
+}