peter/ekster
1
0
Fork 0
Code Issues 3 Pull Requests Projects 3 Releases 1 Wiki Activity

Compare commits

base: peter:3a43844e937f05507008ce9d0d6bc0b93ae74226
Branches Tags
peter:master
peter:wip-jwt-channels
peter:cleanup
peter:1.0.0-rc.1
peter:0.8.4
peter:0.8.3
peter:0.8.2
peter:0.8.0
peter:0.4.0
peter:0.3.0
..
compare: peter:4f6ea0efb297d6ca7efdc9ed9ff4422b3d564a6e
Branches Tags
peter:master
peter:wip-jwt-channels
peter:cleanup
peter:1.0.0-rc.1
peter:0.8.4
peter:0.8.3
peter:0.8.2
peter:0.8.0
peter:0.4.0
peter:0.3.0

No commits in common. "3a43844e937f05507008ce9d0d6bc0b93ae74226" and "4f6ea0efb297d6ca7efdc9ed9ff4422b3d564a6e" have entirely different histories.
3a43844e93 ... 4f6ea0efb2

7 changed files with 181 additions and 124 deletions
Show Stats Expand all files Collapse all files

2
cmd/eksterd/db/migrations/000002_create_feeds_table.up.sql
View File

@ -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

1
cmd/eksterd/db/migrations/000006_create_subscriptions_table.down.sql
View File

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

12
cmd/eksterd/db/migrations/000006_create_subscriptions_table.up.sql
View File

@ -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
);

208
cmd/eksterd/hubbackend.go
View File

@ -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
URL string
Callback string
Hub string
Secret string
LeaseSeconds int64
ResubscribeAt time.Time
ID int64 `redis:"id"`
Channel string `redis:"channel"`
URL string `redis:"url"`
Callback string `redis:"callback"`
Hub string `redis:"hub"`
Secret string `redis:"secret"`
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
var secret string
err := db.QueryRow(
`select "subscription_secret" from "subscriptions" where "id" = $1`,
id,
).Scan(&secret)
conn := h.pool.Get()
defer conn.Close()
secret, err := redis.String(conn.Do("HGET", fmt.Sprintf("feed:%d", id), "secret"))
if err != nil {
return ""
}
return secret
}
func (h *hubIncomingBackend) CreateFeed(topic string) (int64, error) {
db := h.backend.database
func (h *hubIncomingBackend) CreateFeed(topic string, channel string) (int64, error) {
conn := h.pool.Get()
defer conn.Close()
secret := util.RandStringBytes(32)
urlSecret := util.RandStringBytes(32)
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)
// TODO(peter): check if topic already is registered
id, err := redis.Int64(conn.Do("INCR", "feed:next_id"))
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 {
db := h.backend.database
var (
topic string
channel string
feedID string
)
// Process all channels that contains this feed
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,
)
func (h *hubIncomingBackend) UpdateFeed(feedID int64, contentType string, body io.Reader) error {
conn := h.pool.Get()
defer conn.Close()
log.Printf("updating feed %d", feedID)
u, err := redis.String(conn.Do("HGET", fmt.Sprintf("feed:%d", feedID), "url"))
if err != nil {
return err
}
channel, err := redis.String(conn.Do("HGET", fmt.Sprintf("feed:%d", feedID), "channel"))
if err != nil {
return err
}
for rows.Next() {
err = rows.Scan(&topic, channel, feedID)
if err != nil {
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)
}
// FIXME: feed id for incoming websub content
log.Printf("Updating feed %d - %s %s\n", feedID, u, channel)
err = h.backend.ProcessContent(channel, fmt.Sprintf("incoming:%d", feedID), u, 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 {
db := h.backend.database
_, err := db.Exec("update subscriptions set lease_seconds = $1, resubscribe_at = now() + $1 * interval '1' second where id = $2", leaseSeconds, subscriptionID)
return err
func (h *hubIncomingBackend) FeedSetLeaseSeconds(feedID int64, leaseSeconds int64) error {
conn := h.pool.Get()
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
var feeds []Feed
conn := h.pool.Get()
defer conn.Close()
feeds := []Feed{}
rows, err := db.Query(`
select s.id, c.uid, topic, hub, callback, subscription_secret, lease_seconds, resubscribe_at
from subscriptions s
inner join feeds f on f.url = s.topic
inner join channels c on c.id = f.channel_id
`)
// FIXME(peter): replace with set of currently checked feeds
feedKeys, err := redis.Strings(conn.Do("KEYS", "feed:*"))
if err != nil {
return nil, errors.Wrap(err, "could not get feeds from backend")
}
for rows.Next() {
for _, feedKey := range feedKeys {
var feed Feed
err = rows.Scan(
&feed.ID,
&feed.URL,
&feed.Hub,
&feed.Callback,
&feed.Secret,
&feed.LeaseSeconds,
&feed.ResubscribeAt,
)
values, err := redis.Values(conn.Do("HGETALL", feedKey))
if err != nil {
log.Println("Feeds: scan subscriptions:", err)
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)
}

62
cmd/eksterd/memory.go
View File

@ -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 err != nil {
if err == sql.ErrNoRows {
return microsub.Feed{}, fmt.Errorf("channel does not exist: %w", err)
if row := b.database.QueryRow(`SELECT "id" FROM "channels" WHERE "uid" = $1`, uid); row != nil {
err := row.Scan(&channelID)
if err != nil {
log.Fatal(err)
}
return microsub.Feed{}, err
}
var feedID int
err = b.database.QueryRow(
row := b.database.QueryRow(
`INSERT INTO "feeds" ("channel_id", "url") VALUES ($1, $2) RETURNING "id"`,
channelID,
feed.URL,
).Scan(&feedID)
if err != nil {
return feed, err
)
if row == nil {
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)
}

1
go.mod
View File

@ -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

19
pkg/timeline/postgres.go
View File

@ -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
optFeedID.Int64 = 0
} else {
optFeedID.Valid = true
optFeedID.Int64 = feedID
}
feedID, err := strconv.ParseInt(item.Source.ID, 10, 64)
if err != nil {
return false, fmt.Errorf("ERROR: item.Source.ID is not an integer %q: %w", item.Source.ID, err)
}
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)
}