cmd/eksterd/db/migrations/000002_create_feeds_table.up.sql

@@ -1,6 +1,6 @@
 CREATE TABLE "feeds" (
      "id" int primary key generated always as identity,
-     "channel_id" int references "channels" (id) on update cascade on delete cascade,
+     "channel_id" int references "channels" on update cascade on delete cascade,
      "url" varchar(512) not null unique,
      "created_at" timestamptz DEFAULT current_timestamp,
      "updated_at" timestamptz

cmd/eksterd/db/migrations/000006_create_subscriptions_table.down.sql

@@ -1 +0,0 @@
-DROP TABLE "subscriptions";

cmd/eksterd/db/migrations/000006_create_subscriptions_table.up.sql

@@ -1,12 +0,0 @@
-CREATE TABLE "subscriptions" (
-     "id" int primary key generated always as identity,
-     "topic" varchar(1024) not null references "feeds" ("url") on update cascade on delete cascade,
-     "hub" varchar(1024) null,
-     "callback" varchar(1024) null,
-     "subscription_secret" varchar(32) not null,
-     "url_secret" varchar(32) not null,
-     "lease_seconds" int not null,
-     "created_at" timestamptz DEFAULT current_timestamp,
-     "updated_at" timestamptz,
-     "resubscribe_at" timestamptz
-);

cmd/eksterd/hubbackend.go

@@ -6,8 +6,12 @@ import (
 	"io"
 	"log"
 	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
 	"time"
 
+	"github.com/pkg/errors"
 	"p83.nl/go/ekster/pkg/util"
 	"p83.nl/go/ekster/pkg/websub"
 
@@ -19,8 +23,9 @@ const LeaseSeconds = 24 * 60 * 60
 
 // HubBackend handles information for the incoming handler
 type HubBackend interface {
+	GetFeeds() []Feed // Deprecated
 	Feeds() ([]Feed, error)
-	CreateFeed(url string) (int64, error)
+	CreateFeed(url, channel string) (int64, error)
 	GetSecret(feedID int64) string
 	UpdateFeed(feedID int64, contentType string, body io.Reader) error
 	FeedSetLeaseSeconds(feedID int64, leaseSeconds int64) error
@@ -35,13 +40,14 @@ type hubIncomingBackend struct {
 
 // Feed contains information about the feed subscriptions
 type Feed struct {
-	ID            int64
+	ID            int64  `redis:"id"`
-	URL           string
+	Channel       string `redis:"channel"`
-	Callback      string
+	URL           string `redis:"url"`
-	Hub           string
+	Callback      string `redis:"callback"`
-	Secret        string
+	Hub           string `redis:"hub"`
-	LeaseSeconds  int64
+	Secret        string `redis:"secret"`
-	ResubscribeAt time.Time
+	LeaseSeconds  int64  `redis:"lease_seconds"`
+	ResubscribeAt int64  `redis:"resubscribe_at"`
 }
 
 var (
@@ -53,34 +59,29 @@ func init() {
 }
 
 func (h *hubIncomingBackend) GetSecret(id int64) string {
-	db := h.backend.database
+	conn := h.pool.Get()
-	var secret string
+	defer conn.Close()
-	err := db.QueryRow(
+	secret, err := redis.String(conn.Do("HGET", fmt.Sprintf("feed:%d", id), "secret"))
-		`select "subscription_secret" from "subscriptions" where "id" = $1`,
-		id,
-	).Scan(&secret)
 	if err != nil {
 		return ""
 	}
 	return secret
 }
 
-func (h *hubIncomingBackend) CreateFeed(topic string) (int64, error) {
+func (h *hubIncomingBackend) CreateFeed(topic string, channel string) (int64, error) {
-	db := h.backend.database
+	conn := h.pool.Get()
+	defer conn.Close()
 
-	secret := util.RandStringBytes(32)
+	// TODO(peter): check if topic already is registered
-	urlSecret := util.RandStringBytes(32)
+	id, err := redis.Int64(conn.Do("INCR", "feed:next_id"))
-
-	var subscriptionID int
-	err := db.QueryRow(`
-INSERT INTO "subscriptions" ("topic","subscription_secret", "url_secret", "lease_seconds", "created_at")
-VALUES ($1, $2, $3, $4, DEFAULT) RETURNING "id"`, topic, secret, urlSecret, 60*60*24*7).Scan(&subscriptionID)
 	if err != nil {
 		return 0, err
 	}
-	if err != nil {
+
-		return 0, fmt.Errorf("insert into subscriptions: %w", err)
+	conn.Do("HSET", fmt.Sprintf("feed:%d", id), "url", topic)
-	}
+	conn.Do("HSET", fmt.Sprintf("feed:%d", id), "channel", channel)
+	secret := util.RandStringBytes(16)
+	conn.Do("HSET", fmt.Sprintf("feed:%d", id), "secret", secret)
 
 	client := &http.Client{}
 
@@ -90,95 +91,136 @@ VALUES ($1, $2, $3, $4, DEFAULT) RETURNING "id"`, topic, secret, urlSecret, 60*6
 		return 0, err
 	}
 
-	callbackURL := fmt.Sprintf("%s/incoming/%d", h.baseURL, subscriptionID)
+	callbackURL := fmt.Sprintf("%s/incoming/%d", h.baseURL, id)
 
 	log.Printf("WebSub Hub URL found for topic=%q hub=%q callback=%q\n", topic, hubURL, callbackURL)
 
 	if err == nil && hubURL != "" {
-		_, err := db.Exec(`UPDATE subscriptions SET hub = $1, callback = $2 WHERE id = $3`, subscriptionID, hubURL, callbackURL)
+		args := redis.Args{}.Add(fmt.Sprintf("feed:%d", id), "hub", hubURL, "callback", callbackURL)
+		_, err = conn.Do("HMSET", args...)
 		if err != nil {
-			return 0, fmt.Errorf("save hub and callback: %w", err)
+			return 0, errors.Wrap(err, "could not write to redis backend")
 		}
 	} else {
-		return int64(subscriptionID), nil
+		return id, nil
 	}
 
 	err = websub.Subscribe(client, hubURL, topic, callbackURL, secret, 24*3600)
 	if err != nil {
-		return 0, fmt.Errorf("subscribe: %w", err)
+		return 0, err
 	}
 
-	return int64(subscriptionID), nil
+	return id, nil
 }
 
-func (h *hubIncomingBackend) UpdateFeed(subscriptionID int64, contentType string, body io.Reader) error {
+func (h *hubIncomingBackend) UpdateFeed(feedID int64, contentType string, body io.Reader) error {
-	db := h.backend.database
+	conn := h.pool.Get()
-	var (
+	defer conn.Close()
-		topic   string
+	log.Printf("updating feed %d", feedID)
-		channel string
+	u, err := redis.String(conn.Do("HGET", fmt.Sprintf("feed:%d", feedID), "url"))
-		feedID  string
+	if err != nil {
-	)
+		return err
-
+	}
-	// Process all channels that contains this feed
+	channel, err := redis.String(conn.Do("HGET", fmt.Sprintf("feed:%d", feedID), "channel"))
-	rows, err := db.Query(
-		`select topic, c.uid, f.id from subscriptions s inner join feeds f on f.url = s.topic inner join channels c on c.id = f.channel_id where s.id = $1`,
-		subscriptionID,
-	)
 	if err != nil {
 		return err
 	}
 
-	for rows.Next() {
+	// FIXME: feed id for incoming websub content
-		err = rows.Scan(&topic, channel, feedID)
+	log.Printf("Updating feed %d - %s %s\n", feedID, u, channel)
-		if err != nil {
+	err = h.backend.ProcessContent(channel, fmt.Sprintf("incoming:%d", feedID), u, contentType, body)
-			log.Println(err)
-			continue
-		}
-
-		log.Printf("Updating feed %s %q in %q\n", feedID, topic, channel)
-		err = h.backend.ProcessContent(channel, feedID, topic, contentType, body)
 	if err != nil {
 		log.Printf("could not process content for channel %s: %s", channel, err)
 	}
-	}
 
 	return err
 }
 
-func (h *hubIncomingBackend) FeedSetLeaseSeconds(subscriptionID int64, leaseSeconds int64) error {
+func (h *hubIncomingBackend) FeedSetLeaseSeconds(feedID int64, leaseSeconds int64) error {
-	db := h.backend.database
+	conn := h.pool.Get()
-	_, err := db.Exec("update subscriptions set lease_seconds = $1, resubscribe_at = now() + $1 * interval '1' second where id = $2", leaseSeconds, subscriptionID)
+	defer conn.Close()
+	log.Printf("updating feed %d lease_seconds", feedID)
+
+	args := redis.Args{}.Add(fmt.Sprintf("feed:%d", feedID), "lease_seconds", leaseSeconds, "resubscribe_at", time.Now().Add(time.Duration(60*(leaseSeconds-15))*time.Second).Unix())
+	_, err := conn.Do("HMSET", args...)
+	if err != nil {
+		log.Println(err)
 		return err
 	}
 
+	return nil
+}
+
+// GetFeeds is deprecated, use Feeds instead
+func (h *hubIncomingBackend) GetFeeds() []Feed {
+	log.Println("GetFeeds called, consider replacing with Feeds")
+	feeds, err := h.Feeds()
+	if err != nil {
+		log.Printf("Feeds returned an error: %v", err)
+	}
+	return feeds
+}
+
 // Feeds returns a list of subscribed feeds
 func (h *hubIncomingBackend) Feeds() ([]Feed, error) {
-	db := h.backend.database
+	conn := h.pool.Get()
-	var feeds []Feed
+	defer conn.Close()
+	feeds := []Feed{}
 
-	rows, err := db.Query(`
+	// FIXME(peter): replace with set of currently checked feeds
-		select s.id, c.uid, topic, hub, callback, subscription_secret, lease_seconds, resubscribe_at
+	feedKeys, err := redis.Strings(conn.Do("KEYS", "feed:*"))
-		from subscriptions s
-		inner join feeds f on f.url = s.topic
-		inner join channels c on c.id = f.channel_id
-	`)
-
-	for rows.Next() {
-		var feed Feed
-
-		err = rows.Scan(
-			&feed.ID,
-			&feed.URL,
-			&feed.Hub,
-			&feed.Callback,
-			&feed.Secret,
-			&feed.LeaseSeconds,
-			&feed.ResubscribeAt,
-		)
 	if err != nil {
-			log.Println("Feeds: scan subscriptions:", err)
+		return nil, errors.Wrap(err, "could not get feeds from backend")
+	}
+
+	for _, feedKey := range feedKeys {
+		var feed Feed
+		values, err := redis.Values(conn.Do("HGETALL", feedKey))
+		if err != nil {
+			log.Printf("could not get feed info for key %s: %v", feedKey, err)
 			continue
 		}
+
+		err = redis.ScanStruct(values, &feed)
+		if err != nil {
+			log.Printf("could not scan struct for key %s: %v", feedKey, err)
+			continue
+		}
+
+		// Add feed id
+		if feed.ID == 0 {
+			parts := strings.Split(feedKey, ":")
+			if len(parts) == 2 {
+				feed.ID, _ = strconv.ParseInt(parts[1], 10, 64)
+				_, err = conn.Do("HSET", feedKey, "id", feed.ID)
+				if err != nil {
+					log.Printf("could not save id for %s: %v", feedKey, err)
+				}
+			}
+		}
+
+		// Fix the callback url
+		callbackURL, err := url.Parse(feed.Callback)
+		if err != nil || !callbackURL.IsAbs() {
+			if err != nil {
+				log.Printf("could not parse callback url %q: %v", callbackURL, err)
+			} else {
+				log.Printf("url is relative; replace with absolute url: %q", callbackURL)
+			}
+
+			feed.Callback = fmt.Sprintf("%s/incoming/%d", h.baseURL, feed.ID)
+			_, err = conn.Do("HSET", feedKey, "callback", feed.Callback)
+			if err != nil {
+				log.Printf("could not save id for %s: %v", feedKey, err)
+			}
+		}
+
+		// Skip feeds without a Hub
+		if feed.Hub == "" {
+			continue
+		}
+
+		log.Printf("Websub feed: %#v\n", feed)
 		feeds = append(feeds, feed)
 	}
 
@@ -203,13 +245,11 @@ func (h *hubIncomingBackend) run() error {
 
 				feeds, err := h.Feeds()
 				if err != nil {
-					log.Println("Feeds failed:", err)
-					continue
 				}
 
 				for _, feed := range feeds {
 					log.Printf("Looking at %s\n", feed.URL)
-					if time.Now().After(feed.ResubscribeAt) {
+					if feed.ResubscribeAt == 0 || time.Now().After(time.Unix(feed.ResubscribeAt, 0)) {
 						if feed.Callback == "" {
 							feed.Callback = fmt.Sprintf("%s/incoming/%d", h.baseURL, feed.ID)
 						}

cmd/eksterd/memory.go

@@ -312,23 +312,23 @@ func (b *memoryBackend) FollowURL(uid string, url string) (microsub.Feed, error)
 	feed := microsub.Feed{Type: "feed", URL: url}
 
 	var channelID int
-	err := b.database.QueryRow(`SELECT "id" FROM "channels" WHERE "uid" = $1`, uid).Scan(&channelID)
+	if row := b.database.QueryRow(`SELECT "id" FROM "channels" WHERE "uid" = $1`, uid); row != nil {
+		err := row.Scan(&channelID)
 		if err != nil {
-		if err == sql.ErrNoRows {
+			log.Fatal(err)
-			return microsub.Feed{}, fmt.Errorf("channel does not exist: %w", err)
 		}
-		return microsub.Feed{}, err
 	}
 
-	var feedID int
+	row := b.database.QueryRow(
-	err = b.database.QueryRow(
 		`INSERT INTO "feeds" ("channel_id", "url") VALUES ($1, $2) RETURNING "id"`,
 		channelID,
 		feed.URL,
-	).Scan(&feedID)
+	)
-	if err != nil {
+	if row == nil {
-		return feed, err
+		return microsub.Feed{}, fmt.Errorf("no feed_id")
 	}
+	var feedID int
+	_ = row.Scan(&feedID)
 
 	resp, err := b.Fetch3(uid, feed.URL)
 	if err != nil {
@@ -804,3 +804,45 @@ func (b *memoryBackend) getTimeline(channel string) timeline.Backend {
 	}
 	return tl
 }
+
+func (b *memoryBackend) createChannel(name string) microsub.Channel {
+	uid := fmt.Sprintf("%012d", b.NextUID)
+	channel := microsub.Channel{
+		UID:    uid,
+		Name:   name,
+		Unread: microsub.Unread{Type: microsub.UnreadCount},
+	}
+	return channel
+}
+
+func (b *memoryBackend) fetchChannel(name string) (microsub.Channel, bool) {
+	b.lock.RLock()
+	defer b.lock.RUnlock()
+
+	for _, c := range b.Channels {
+		if c.Name == name {
+			return c, true
+		}
+	}
+
+	return microsub.Channel{}, false
+}
+
+func (b *memoryBackend) setChannel(channel microsub.Channel) {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+
+	b.Channels[channel.UID] = channel
+	b.Feeds[channel.UID] = []microsub.Feed{}
+	b.NextUID++
+}
+
+func updateChannelInRedis(conn redis.Conn, uid string, prio int) {
+	conn.Do("SADD", "channels", uid)
+	conn.Do("SETNX", "channel_sortorder_"+uid, prio)
+}
+
+func removeChannelFromRedis(conn redis.Conn, uid string) {
+	conn.Do("SREM", "channels", uid)
+	conn.Do("DEL", "channel_sortorder_"+uid)
+}

go.mod

@@ -5,7 +5,6 @@ go 1.16
 require (
 	github.com/axgle/mahonia v0.0.0-20180208002826-3358181d7394
 	github.com/blevesearch/bleve/v2 v2.0.3
-	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/gilliek/go-opml v1.0.0
 	github.com/golang-migrate/migrate/v4 v4.15.1
 	github.com/gomodule/redigo v1.8.2

pkg/timeline/postgres.go

@@ -212,31 +212,20 @@ func (p *postgresStream) AddItem(item microsub.Item) (bool, error) {
 		t = t2
 	}
 	if item.UID == "" {
-		// FIXME: This won't work when we receive the item multiple times
 		h := sha256.Sum256([]byte(fmt.Sprintf("%s:%d", p.channel, time.Now().UnixNano())))
 		item.UID = hex.EncodeToString(h[:])
 	}
 
-	var optFeedID sql.NullInt64
-	if item.Source == nil || item.Source.ID == "" {
-		optFeedID.Valid = false
-		optFeedID.Int64 = 0
-	} else {
 	feedID, err := strconv.ParseInt(item.Source.ID, 10, 64)
 	if err != nil {
-			optFeedID.Valid = false
+		return false, fmt.Errorf("ERROR: item.Source.ID is not an integer %q: %w", item.Source.ID, err)
-			optFeedID.Int64 = 0
-		} else {
-			optFeedID.Valid = true
-			optFeedID.Int64 = feedID
-		}
 	}
 
 	result, err := conn.ExecContext(context.Background(), `
 INSERT INTO "items" ("channel_id", "feed_id", "uid", "data", "published_at", "created_at")
 VALUES ($1, $2, $3, $4, $5, DEFAULT)
 ON CONFLICT ON CONSTRAINT "items_uid_key" DO NOTHING
-`, p.channelID, optFeedID, item.UID, &item, t)
+`, p.channelID, feedID, item.UID, &item, t)
 	if err != nil {
 		return false, fmt.Errorf("insert item: %w", err)
 	}