cmd/eksterd/app.go

@@ -59,12 +59,7 @@ func NewApp(options AppOptions) (*App, error) {
 	app.backend.hubIncomingBackend.baseURL = options.BaseURL
 	app.backend.hubIncomingBackend.backend = app.backend
 
-	app.hubBackend = &hubIncomingBackend{
-		backend:  app.backend,
-		baseURL:  options.BaseURL,
-		pool:     options.pool,
-		database: options.database,
-	}
+	app.hubBackend = &hubIncomingBackend{backend: app.backend, baseURL: options.BaseURL, pool: options.pool}
 	app.backend.hubIncomingBackend = *app.hubBackend
 
 	http.Handle("/micropub", &micropubHandler{
@@ -82,8 +77,7 @@ func NewApp(options AppOptions) (*App, error) {
 	http.Handle("/microsub", handler)
 
 	http.Handle("/incoming/", &incomingHandler{
-		Backend:   app.hubBackend,
-		Processor: app.backend,
+		Backend: app.hubBackend,
 	})
 
 	if !options.Headless {

cmd/eksterd/hubbackend.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"database/sql"
 	"expvar"
 	"fmt"
 	"io"
@@ -23,16 +22,15 @@ type HubBackend interface {
 	Feeds() ([]Feed, error)
 	CreateFeed(url string) (int64, error)
 	GetSecret(feedID int64) string
-	UpdateFeed(processor ContentProcessor, feedID int64, contentType string, body io.Reader) error
+	UpdateFeed(feedID int64, contentType string, body io.Reader) error
 	FeedSetLeaseSeconds(feedID int64, leaseSeconds int64) error
 	Subscribe(feed *Feed) error
 }
 
 type hubIncomingBackend struct {
-	backend  *memoryBackend
-	baseURL  string
-	pool     *redis.Pool
-	database *sql.DB
+	backend *memoryBackend
+	baseURL string
+	pool    *redis.Pool
 }
 
 // Feed contains information about the feed subscriptions
@@ -55,7 +53,7 @@ func init() {
 }
 
 func (h *hubIncomingBackend) GetSecret(id int64) string {
-	db := h.database
+	db := h.backend.database
 	var secret string
 	err := db.QueryRow(
 		`select "subscription_secret" from "subscriptions" where "id" = $1`,
@@ -69,7 +67,7 @@ func (h *hubIncomingBackend) GetSecret(id int64) string {
 
 func (h *hubIncomingBackend) CreateFeed(topic string) (int64, error) {
 	log.Println("CreateFeed", topic)
-	db := h.database
+	db := h.backend.database
 
 	secret := util.RandStringBytes(32)
 	urlSecret := util.RandStringBytes(32)
@@ -114,10 +112,10 @@ VALUES ($1, $2, $3, $4, DEFAULT) RETURNING "id"`, topic, secret, urlSecret, 60*6
 	return int64(subscriptionID), nil
 }
 
-func (h *hubIncomingBackend) UpdateFeed(processor ContentProcessor, subscriptionID int64, contentType string, body io.Reader) error {
+func (h *hubIncomingBackend) UpdateFeed(subscriptionID int64, contentType string, body io.Reader) error {
 	log.Println("UpdateFeed", subscriptionID)
 
-	db := h.database
+	db := h.backend.database
 	var (
 		topic   string
 		channel string
@@ -141,7 +139,7 @@ func (h *hubIncomingBackend) UpdateFeed(processor ContentProcessor, subscription
 		}
 
 		log.Printf("Updating feed %s %q in %q\n", feedID, topic, channel)
-		err = processor.ProcessContent(channel, feedID, topic, contentType, body)
+		err = h.backend.ProcessContent(channel, feedID, topic, contentType, body)
 		if err != nil {
 			log.Printf("could not process content for channel %s: %s", channel, err)
 		}
@@ -151,7 +149,7 @@ func (h *hubIncomingBackend) UpdateFeed(processor ContentProcessor, subscription
 }
 
 func (h *hubIncomingBackend) FeedSetLeaseSeconds(subscriptionID int64, leaseSeconds int64) error {
-	db := h.database
+	db := h.backend.database
 	_, err := db.Exec(`
 update subscriptions
 set lease_seconds = $1,
@@ -163,7 +161,7 @@ where id = $3
 
 // Feeds returns a list of subscribed feeds
 func (h *hubIncomingBackend) Feeds() ([]Feed, error) {
-	db := h.database
+	db := h.backend.database
 	var feeds []Feed
 
 	rows, err := db.Query(`

cmd/eksterd/incoming.go

@@ -13,8 +13,7 @@ import (
 )
 
 type incomingHandler struct {
-	Backend   HubBackend
-	Processor ContentProcessor
+	Backend HubBackend
 }
 
 var (
@@ -95,7 +94,7 @@ func (h *incomingHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	ct := r.Header.Get("Content-Type")
-	err = h.Backend.UpdateFeed(h.Processor, feed, ct, bytes.NewBuffer(feedContent))
+	err = h.Backend.UpdateFeed(feed, ct, bytes.NewBuffer(feedContent))
 	if err != nil {
 		http.Error(w, fmt.Sprintf("could not update feed: %s (%s)", ct, err), 400)
 		return

cmd/eksterd/memory.go

@@ -573,11 +573,6 @@ func ProcessSourcedItems(fetcher fetch.Fetcher, fetchURL, contentType string, bo
 	return items, nil
 }
 
-// ContentProcessor processes content for a channel and feed
-type ContentProcessor interface {
-	ProcessContent(channel, feedID, fetchURL, contentType string, body io.Reader) error
-}
-
 func (b *memoryBackend) ProcessContent(channel, feedID, fetchURL, contentType string, body io.Reader) error {
 	cachingFetch := WithCaching(b.pool, fetch.FetcherFunc(Fetch2))