autogits/gitea-events-rabbitmq-publisher/main.go

package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"

	"src.opensuse.org/autogits/common"
)

const (
	ListenAddrDef = "[::1]:8002"
	AppName       = "rabbitmq-forwarder"
)

var DebugMode bool

func connectToRabbitMQ() {
	host := os.Getenv("RABBITMQ_HOST")
	username := os.Getenv("RABBITMQ_USERNAME")
	password := os.Getenv("RABBITMQ_PASSWORD")

	if len(host) == 0 || len(username) == 0 || len(password) == 0 {
		fmt.Println("Missing RABBITMQ_HOST, RABBITMQ_USERNAME, RABBITMQ_PASSWORD")
		os.Exit(1)
	}

	go ConnectToExchangeForPublish(host, username, password)
}

var id int

func dumpUnhandledData(reqType string, data []byte) {
	id++
	os.WriteFile("/tmp/unhandled-json-"+reqType+"-"+fmt.Sprint(id)+".json", data, 0600)
}
func main() {
	var listenAddr string

	flag.BoolVar(&DebugMode, "debug", false, "enables debugging messages")
	flag.StringVar(&listenAddr, "listen", ListenAddrDef, "HTTP listen socket address for webhook events")
	flag.Parse()

	connectToRabbitMQ()

	http.HandleFunc("POST /test", func(res http.ResponseWriter, req *http.Request) {
		if len(req.Header.Get("Content-Type")) == 0 ||
			req.Header["Content-Type"][0] != "application/json" ||
			req.Method != "POST" {

			res.WriteHeader(http.StatusInternalServerError)
			return
		}

		hdr := req.Header[common.GiteaRequestHeader]
		if len(hdr) != 1 {
			res.WriteHeader(http.StatusInternalServerError)
			log.Printf("Multiple Gitea headers received. %#v\n", hdr)
			if DebugMode {
				log.Println(req.Header)
			}
			return
		}
		reqType := hdr[0]
		data, err := io.ReadAll(req.Body)
		if err != nil {
			errorStr := fmt.Sprintf("error reading hook info: %v", err)
			res.Header().Add("Content-Type", "plain/text")
			res.Write([]byte(errorStr))
			res.WriteHeader(http.StatusBadRequest)

			if DebugMode {
				log.Printf(errorStr)
			}
		}

		if !json.Valid(data) {
			if DebugMode {
				log.Println("send invalid json request")
			}
			res.WriteHeader(http.StatusBadRequest)
		}

		org := common.Organization{}
		extraAction := ""

		switch reqType {
		case "create", "delete":
			create := common.CreateWebhookEvent{}
			if err = json.Unmarshal(data, &create); err != nil {
				res.WriteHeader(http.StatusBadRequest)
				if DebugMode {
					dumpUnhandledData(reqType, data)
				}
				return
			}
			org = *create.Repository.Owner

		case "fork":
			fork := common.ForkWebhookEvent{}
			if err = json.Unmarshal(data, &fork); err != nil {
				res.WriteHeader(http.StatusBadRequest)
				if DebugMode {
					dumpUnhandledData(reqType, data)
				}
				return
			}
			org = *fork.Forkee.Owner
		case "push":
			push := common.PushRequest{}
			if err = json.Unmarshal(data, &push); err != nil {
				res.WriteHeader(http.StatusBadRequest)
				if DebugMode {
					dumpUnhandledData(reqType, data)
				}
				return
			}
			org = *push.Repository.Owner
		case "repository":
			repoAction := common.RepositoryAction{}
			if err = json.Unmarshal(data, &repoAction); err != nil {
				res.WriteHeader(http.StatusBadRequest)
				if DebugMode {
					dumpUnhandledData(reqType, data)
				}
				return
			}
			switch repoAction.Action {
			case "created", "deleted":
				break
			default:
				res.WriteHeader(http.StatusBadRequest)
				if DebugMode {
					dumpUnhandledData(reqType, data)
				}
			}
			org = repoAction.Organization
			extraAction = repoAction.Action
		case "release":
			release := common.ReleaseWebhookEvent{}
			if err = json.Unmarshal(data, &release); err != nil {
				res.WriteHeader(http.StatusBadRequest)
				if DebugMode {
					dumpUnhandledData(reqType, data)
				}
				return
			}
			switch release.Action {
			case "published", "updated", "deleted":
				break
			default:
				res.WriteHeader(http.StatusBadRequest)
				if DebugMode {
					dumpUnhandledData(reqType, data)
				}
				return
			}

			org = *release.Repository.Owner
			extraAction = release.Action
		default:
			res.WriteHeader(http.StatusBadRequest)
			if DebugMode {
				dumpUnhandledData(reqType, data)
			}
		}

		//		err = PublishMessage(repo.Owner.Username, reqType, data)
		// write to file for review
		id++
		os.WriteFile("test_data."+fmt.Sprint(id), data, 0644)
		log.Println(id, org.Username, extraAction)

		/*
			err = PublishMessage(org.Name, reqType, extraAction, data)
			if err != nil {
				errorStr := fmt.Sprintf("hook (%s) processing error: %v\n", reqType, err)
				res.Header().Add("Content-Type", "plain/text")
				res.Write([]byte(errorStr))
				res.WriteHeader(http.StatusBadRequest)

				if DebugMode {
					log.Println(errorStr)
				}
			} else {*/
		res.WriteHeader(http.StatusOK)
		//		}
	})

	http.HandleFunc("POST /rabbitmq-forwarder/{Org}", func(res http.ResponseWriter, req *http.Request) {
		if len(req.Header.Get("Content-Type")) == 0 ||
			req.Header["Content-Type"][0] != "application/json" ||
			req.Method != "POST" {

			res.WriteHeader(http.StatusInternalServerError)
			return
		}

		hdr := req.Header[common.GiteaRequestHeader]
		if len(hdr) != 1 {
			res.WriteHeader(http.StatusInternalServerError)
			log.Printf("Multiple Gitea headers received. %#v\n", hdr)
			if DebugMode {
				log.Println(req.Header)
			}
			return
		}
		reqType := hdr[0]
		data, err := io.ReadAll(req.Body)
		if err != nil {
			errorStr := fmt.Sprintf("error reading hook info: %v", err)
			res.Header().Add("Content-Type", "plain/text")
			res.Write([]byte(errorStr))
			res.WriteHeader(http.StatusBadRequest)

			if DebugMode {
				log.Printf(errorStr)
			}
		}

		err = PublishMessage(req.PathValue("Org"), reqType, data)
		if err != nil {
			errorStr := fmt.Sprintf("hook (%s) processing error: %v\n", reqType, err)
			res.Header().Add("Content-Type", "plain/text")
			res.Write([]byte(errorStr))
			res.WriteHeader(http.StatusBadRequest)

			if DebugMode {
				log.Println(errorStr)
			}
		} else {
			res.WriteHeader(http.StatusOK)
		}
	})

	log.Fatal(http.ListenAndServe(listenAddr, nil))
}