khatru/rss-bridge/main.go

package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/cockroachdb/pebble"
	"github.com/fiatjaf/relayer"
	"github.com/kelseyhightower/envconfig"
	"github.com/nbd-wtf/go-nostr"
	"golang.org/x/exp/slices"
)

var relay = &Relay{
	updates: make(chan nostr.Event),
}

type Relay struct {
	Secret string `envconfig:"SECRET" required:"true"`

	updates     chan nostr.Event
	lastEmitted sync.Map
	db          *pebble.DB
}

func (relay *Relay) Name() string {
	return "relayer-rss-bridge"
}

func (r *Relay) OnInitialized(s *relayer.Server) {
	s.Router().Path("/").HandlerFunc(handleWebpage)
	s.Router().Path("/create").HandlerFunc(handleCreateFeed)
}

func (relay *Relay) Init() error {
	err := envconfig.Process("", relay)
	if err != nil {
		return fmt.Errorf("couldn't process envconfig: %w", err)
	}

	if db, err := pebble.Open("db", nil); err != nil {
		log.Fatalf("failed to open db: %v", err)
	} else {
		relay.db = db
	}

	go func() {
		time.Sleep(20 * time.Minute)

		filters := relayer.GetListeningFilters()
		log.Printf("checking for updates; %d filters active", len(filters))

		for _, filter := range filters {
			if filter.Kinds == nil || slices.Contains(filter.Kinds, nostr.KindTextNote) {
				for _, pubkey := range filter.Authors {
					if val, closer, err := relay.db.Get([]byte(pubkey)); err == nil {
						defer closer.Close()

						var entity Entity
						if err := json.Unmarshal(val, &entity); err != nil {
							log.Printf("got invalid json from db at key %s: %v", pubkey, err)
							continue
						}

						feed, err := parseFeed(entity.URL)
						if err != nil {
							log.Printf("failed to parse feed at url %q: %v", entity.URL, err)
							continue
						}

						for _, item := range feed.Items {
							evt := itemToTextNote(pubkey, item)
							last, ok := relay.lastEmitted.Load(entity.URL)
							if !ok || time.Unix(last.(int64), 0).Before(evt.CreatedAt) {
								evt.Sign(entity.PrivateKey)
								relay.updates <- evt
								relay.lastEmitted.Store(entity.URL, last)
							}
						}
					}
				}
			}
		}
	}()

	return nil
}

func (relay *Relay) AcceptEvent(_ *nostr.Event) bool {
	return false
}

func (relay *Relay) Storage() relayer.Storage {
	return store{relay.db}
}

type store struct {
	db *pebble.DB
}

func (b store) Init() error { return nil }
func (b store) SaveEvent(_ *nostr.Event) error {
	return errors.New("blocked: we don't accept any events")
}

func (b store) DeleteEvent(_, _ string) error {
	return errors.New("blocked: we can't delete any events")
}

func (b store) QueryEvents(filter *nostr.Filter) ([]nostr.Event, error) {
	var evts []nostr.Event

	if filter.IDs != nil || len(filter.Tags) > 0 {
		return evts, nil
	}

	for _, pubkey := range filter.Authors {
		if val, closer, err := relay.db.Get([]byte(pubkey)); err == nil {
			defer closer.Close()

			var entity Entity
			if err := json.Unmarshal(val, &entity); err != nil {
				log.Printf("got invalid json from db at key %s: %v", pubkey, err)
				continue
			}

			feed, err := parseFeed(entity.URL)
			if err != nil {
				log.Printf("failed to parse feed at url %q: %v", entity.URL, err)
				continue
			}

			if filter.Kinds == nil || slices.Contains(filter.Kinds, nostr.KindSetMetadata) {
				evt := feedToSetMetadata(pubkey, feed)

				if filter.Since != nil && evt.CreatedAt.Before(*filter.Since) {
					continue
				}
				if filter.Until != nil && evt.CreatedAt.After(*filter.Until) {
					continue
				}

				evt.Sign(entity.PrivateKey)
				evts = append(evts, evt)
			}

			if filter.Kinds == nil || slices.Contains(filter.Kinds, nostr.KindTextNote) {
				var last uint32 = 0
				for _, item := range feed.Items {
					evt := itemToTextNote(pubkey, item)

					if filter.Since != nil && evt.CreatedAt.Before(*filter.Since) {
						continue
					}
					if filter.Until != nil && evt.CreatedAt.After(*filter.Until) {
						continue
					}

					evt.Sign(entity.PrivateKey)

					if evt.CreatedAt.After(time.Unix(int64(last), 0)) {
						last = uint32(evt.CreatedAt.Unix())
					}

					evts = append(evts, evt)
				}

				relay.lastEmitted.Store(entity.URL, last)
			}
		}
	}

	return evts, nil
}

func (relay *Relay) InjectEvents() chan nostr.Event {
	return relay.updates
}

func main() {
	if err := relayer.Start(relay); err != nil {
		log.Fatalf("server terminated: %v", err)
	}
}