gitea-obs/main.go

// Open Build System Gitea trigger webhook
package main

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"net/url"
	"os"
	"os/signal"
	"path"
	"path/filepath"
	"strings"
	"syscall"
	"time"
)

var programName = path.Base(os.Args[0])
var secretToken = ""
var apiClient *http.Client = &http.Client{Timeout: 60 * time.Second}

// We're on a relatively short timeout (10s), just queue the events for later
func handler(w http.ResponseWriter, r *http.Request) {
	data, err := ioutil.ReadAll(r.Body)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	// Presence of the header marks the format of hook data
	event := r.Header.Get("X-Gitea-Event")
	if event == "" {
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	// React to new commits being added
	if event != "push" {
		return
	}

	buf := new(bytes.Buffer)
	if err := json.Indent(buf, data, "", "  "); err != nil {
		log.Printf("req: %s", err)
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	// Mark our files as such, use the filesystem as a simple database
	base := programName + "&" + r.URL.RawQuery
	if err := func() error {
		f, err := os.OpenFile(base+".tmp",
			os.O_WRONLY|os.O_CREATE|os.O_TRUNC|os.O_SYNC, 0644)
		if err != nil {
			return err
		}
		if _, err := f.Write(buf.Bytes()); err != nil {
			f.Close()
			return err
		}
		if err := f.Close(); err != nil {
			return err
		}
		// Ready for processing -- needs to be the last step here
		if err := os.Rename(base+".tmp", base); err != nil {
			return err
		}
		return nil
	}(); err != nil {
		log.Printf("req: %s", err)
		w.WriteHeader(http.StatusInternalServerError)
	}
}

func process(ctx context.Context, values url.Values, data []byte) error {
	pkg := values.Get("package")
	project := values.Get("project")
	token := values.Get("token")
	if token == "" {
		return fmt.Errorf("no token given")
	}

	obs := "https://build.opensuse.org"
	if v := values.Get("obs"); v != "" {
		obs = v
	}

	refs := "refs/heads/master"
	if v := values.Get("refs"); v != "" {
		refs = v
	}

	// No need for the whole "code.gitea.io/sdk/gitea" package so far
	push := struct {
		Secret string `json:"secret"`
		Ref    string `json:"ref"`
	}{}
	if err := json.Unmarshal(data, &push); err != nil {
		return err
	}
	if push.Secret != secretToken {
		return fmt.Errorf("invalid secret: %s", push.Secret)
	}

	// This roughly matches the behaviour of the GitHub service
	matches := false
	for _, ref := range strings.Split(refs, ":") {
		if m, err := filepath.Match(ref, push.Ref); m {
			matches = true
			break
		} else if err != nil {
			return fmt.Errorf("%s: %s", ref, err)
		}
	}

	// Trigger the OBS source service to make it rebuild the package
	if matches {
		uri := obs + "/trigger/runservice"
		if pkg != "" && project != "" {
			escape := url.QueryEscape
			uri += "?package=" + escape(pkg) + "&project=" + escape(project)
		}
		req, err := http.NewRequest("POST", uri, nil)
		if err != nil {
			return err
		}
		req.Header.Set("Authorization", "Token "+token)
		resp, err := apiClient.Do(req.WithContext(ctx))
		if err != nil {
			return err
		}
		defer resp.Body.Close()
	}
	return nil
}

func consume(ctx context.Context) {
	list, err := ioutil.ReadDir(".")
	if err != nil {
		log.Fatalln(err)
	}

	for _, info := range list {
		base := info.Name()
		if strings.HasSuffix(base, ".tmp") || !info.Mode().IsRegular() {
			continue
		}
		data, err := ioutil.ReadFile(base)
		if err != nil {
			log.Println(err)
			continue
		}

		values, err := url.ParseQuery(base)
		if err != nil {
			log.Printf("%s: %s\n", base, err)
			continue
		}
		if _, ok := values[programName]; !ok {
			continue
		}

		if err := process(ctx, values, data); err != nil {
			log.Printf("%s: %s\n", base, err)
			continue
		}
		// XXX minor race condition
		if err := os.Remove(base); err != nil {
			log.Println(err)
			continue
		}
	}
}

func main() {
	if len(os.Args) < 2 || len(os.Args) > 3 {
		fmt.Fprintf(os.Stderr, "Usage: %s LISTEN-ADDR [SECRET]\n", os.Args[0])
		os.Exit(1)
	}

	listenAddr := os.Args[1]
	if len(os.Args) == 3 {
		secretToken = os.Args[2]
	}

	http.HandleFunc("/", handler)
	server := &http.Server{Addr: listenAddr}

	// Run the producer
	errs := make(chan error, 1)
	go func() { errs <- server.ListenAndServe() }()

	// Run the consumer
	ctx, ctxCancel := context.WithCancel(context.Background())
	consumerFinished := make(chan struct{})
	// Process the currently available batch and retry after a few seconds
	go func() {
		defer close(consumerFinished)
		for {
			consume(ctx)
			select {
			case <-time.After(3 * time.Second):
			case <-ctx.Done():
				return
			}
		}
	}()

	// Wait for a termination signal
	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)

	select {
	case <-sigs:
	case err := <-errs:
		log.Println(err)
	}

	// Stop the producer gracefully
	ctxSd, ctxSdCancel := context.WithTimeout(context.Background(), 5)
	if err := server.Shutdown(ctxSd); err != nil {
		log.Println(err)
	}
	ctxSdCancel()

	// Stop the consumer gracefully
	ctxCancel()
	<-consumerFinished
}