hbot/cmd/wurgurboo/cgit.go

/* SPDX-License-Identifier: GPL-2.0
 *
 * Copyright (C) 2021 Jason A. Donenfeld. All Rights Reserved.
 */

package main

import (
	"container/list"
	"encoding/hex"
	"encoding/xml"
	"io"
	"log"
	"net/http"
	"path"
	"sync"
	"time"
)

type CgitCommit struct {
	Text    string `xml:",chardata"`
	Title   string `xml:"title"`
	Updated string `xml:"updated"`
	Author  struct {
		Text  string `xml:",chardata"`
		Name  string `xml:"name"`
		Email string `xml:"email"`
	} `xml:"author"`
	Published string `xml:"published"`
	Link      struct {
		Text string `xml:",chardata"`
		Rel  string `xml:"rel,attr"`
		Type string `xml:"type,attr"`
		Href string `xml:"href,attr"`
	} `xml:"link"`
	ID      string `xml:"id"`
	Content []struct {
		Text string `xml:",chardata"`
		Type string `xml:"type,attr"`
		Div  struct {
			Text  string `xml:",chardata"`
			Xmlns string `xml:"xmlns,attr"`
			Pre   string `xml:"pre"`
		} `xml:"div"`
	} `xml:"content"`
}

type cgitFeed struct {
	XMLName  xml.Name `xml:"feed"`
	Text     string   `xml:",chardata"`
	Xmlns    string   `xml:"xmlns,attr"`
	Title    string   `xml:"title"`
	Subtitle string   `xml:"subtitle"`
	Link     struct {
		Text string `xml:",chardata"`
		Rel  string `xml:"rel,attr"`
		Type string `xml:"type,attr"`
		Href string `xml:"href,attr"`
	} `xml:"link"`
	Entry []CgitCommit `xml:"entry"`
}

type cgitFeedStatus struct {
	repo           string
	title          string
	seenEntries    map[[32]byte]bool
	orderedEntries list.List
}

type CgitFeedMonitorer struct {
	feeds   []*cgitFeedStatus
	updates chan CgitFeedUpdate
	ticker  *time.Ticker
	wg      sync.WaitGroup
	mu      sync.Mutex
}

type CgitFeedUpdate struct {
	RepoTitle string
	Commit    *CgitCommit
}

func NewCgitFeedMonitorer(pollInterval time.Duration) *CgitFeedMonitorer {
	fm := &CgitFeedMonitorer{
		updates: make(chan CgitFeedUpdate, 32),
		ticker:  time.NewTicker(pollInterval),
	}
	fm.wg.Add(1)
	go func() {
		defer fm.wg.Done()
		for range fm.ticker.C {
			fm.mu.Lock()
			for _, feed := range fm.feeds {
				fm.wg.Add(1)
				go func(feed *cgitFeedStatus) {
					defer fm.wg.Done()
					fm.updateFeed(feed, true)
				}(feed)
			}
			fm.mu.Unlock()
		}
	}()
	return fm
}

func (fm *CgitFeedMonitorer) Stop() {
	fm.ticker.Stop()
	fm.wg.Wait()
	close(fm.updates)
}

func (fm *CgitFeedMonitorer) Updates() <-chan CgitFeedUpdate {
	return fm.updates
}

func (fm *CgitFeedMonitorer) AddFeed(repo string) {
	go func() {
		status := &cgitFeedStatus{
			repo:        repo,
			title:       path.Base(repo),
			seenEntries: make(map[[32]byte]bool, 1024),
		}
		fm.updateFeed(status, false)
		fm.mu.Lock()
		fm.feeds = append(fm.feeds, status)
		fm.mu.Unlock()
	}()
}

func (fm *CgitFeedMonitorer) updateFeed(fs *cgitFeedStatus, alert bool) {
	feed, err := fs.fetchFeed()
	if err != nil {
		log.Printf("Unable to fetch commits for %q: %v", fs.title, err)
		return
	}
	for i := len(feed.Entry) - 1; i >= 0; i-- {
		commit := &feed.Entry[i]
		var commitID [32]byte
		if hex.DecodedLen(len(commit.ID)) > len(commitID) {
			continue
		}
		n, err := hex.Decode(commitID[:], []byte(commit.ID))
		if err != nil || n < 20 {
			continue
		}
		if _, ok := fs.seenEntries[commitID]; ok {
			continue
		}
		fs.seenEntries[commitID] = true
		fs.orderedEntries.PushBack(commitID)
		for len(fs.seenEntries) > 1024 {
			first := fs.orderedEntries.Front()
			delete(fs.seenEntries, first.Value.([32]byte))
			fs.orderedEntries.Remove(first)
		}
		if alert {
			fm.updates <- CgitFeedUpdate{fs.title, commit}
		}
	}
}

func (fs *cgitFeedStatus) fetchFeed() (*cgitFeed, error) {
	resp, err := http.Get(fs.repo + "/atom/")
	if err != nil {
		return nil, err
	}
	bytes, err := io.ReadAll(resp.Body)
	resp.Body.Close()
	if err != nil {
		return nil, err
	}
	var feed cgitFeed
	err = xml.Unmarshal(bytes, &feed)
	if err != nil {
		return nil, err
	}
	return &feed, nil
}