WIP: verify JWT

.drone.yml

@@ -1,74 +1,32 @@
 ---
 kind: pipeline
 type: docker
-name: build and test
+name: build
 
 workspace:
   base: /go
   path: src/p83.nl/go/ekster
 
-trigger:
-  event:
-    - push
-
-services:
-  - name: redis
-    image: redis:5
-  - name: database
-    image: postgres:14
-    environment:
-      POSTGRES_DB: ekster_testing
-      POSTGRES_USER: postgres
-      POSTGRES_PASSWORD: simple
-      POSTGRES_HOST_AUTH_METHOD: trust
-
 steps:
   - name: testing
-    image: golang:1.18-alpine
+    image: golang:1.16-alpine
     environment:
       CGO_ENABLED: 0
       GOOS: linux
       GOARCH: amd64
     commands:
-      - go version
       - apk --no-cache add git
       - go get -d -t ./...
-      - go build -buildvcs=false p83.nl/go/ekster/cmd/eksterd
+      - go install honnef.co/go/tools/cmd/staticcheck@latest
+      - go build p83.nl/go/ekster/cmd/eksterd
       - go vet ./...
       - go test -v ./...
-
+      - staticcheck ./...
----
-kind: pipeline
-type: docker
-name: move to production
-
-workspace:
-  base: /go
-  path: src/p83.nl/go/ekster
-
-trigger:
-  event:
-    - promote
-  target:
-    - production
-
-steps:
-  - name: build
-    image: golang:1.18-alpine
-    environment:
-      CGO_ENABLED: 0
-      GOOS: linux
-      GOARCH: amd64
-    commands:
-      - go version
-      - apk --no-cache add git
-      - go get -d -t ./...
-      - go build -buildvcs=false p83.nl/go/ekster/cmd/eksterd
 
   - name: publish-personal
     image: plugins/docker
     depends_on:
-      - build
+      - testing
     settings:
       repo: registry.stuifzandapp.com/microsub-server
       registry: registry.stuifzandapp.com
@@ -76,6 +34,24 @@ steps:
         from_secret: docker_username
       password:
         from_secret: docker_password
+    when:
+      event:
+        - promote
+      target:
+        - production
+
+  # - name: publish-docker
+  #   image: plugins/docker
+  #   depends_on:
+  #     - testing
+  #   settings:
+  #     repo: pstuifzand/ekster
+  #     tags:
+  #       - alpine
+  #     username:
+  #       from_secret: docker_official_username
+  #     password:
+  #       from_secret: docker_official_password
 
   - name: deploy
     image: appleboy/drone-ssh
@@ -90,3 +66,8 @@ steps:
         - cd /home/microsub/microsub
         - docker-compose pull web
         - docker-compose up -d
+    when:
+      event:
+        - promote
+      target:
+        - production

Dockerfile

@@ -3,4 +3,5 @@ RUN apk --no-cache add ca-certificates
 WORKDIR /opt/micropub
 EXPOSE 80
 COPY ./eksterd /app/
+COPY ./templates /app/templates
 ENTRYPOINT ["/app/eksterd"]

cmd/ek/main.go

@@ -242,7 +242,6 @@ func channelID(sub microsub.Microsub, channelNameOrID string) (string, error) {
 		// we encountered an error, so we are not sure if it worked
 		return channelNameOrID, err
 	}
-
 	for _, c := range channels {
 		if c.Name == channelNameOrID {
 			return c.UID, nil

cmd/eksterd/database_test.go

@@ -1,123 +0,0 @@
-/*
- *  Ekster is a microsub server
- *  Copyright (c) 2022 The Ekster authors
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-package main
-
-import (
-	"database/sql"
-	"log"
-	"net/http/httptest"
-	"os"
-	"testing"
-
-	"github.com/gomodule/redigo/redis"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/suite"
-)
-
-type DatabaseSuite struct {
-	suite.Suite
-
-	URL      string
-	Database *sql.DB
-
-	RedisURL string
-	Redis    redis.Conn
-}
-
-func (s *DatabaseSuite) SetupSuite() {
-	db, err := sql.Open("postgres", s.URL)
-	if err != nil {
-		log.Fatal(err)
-	}
-	s.Database = db
-
-	conn, err := redis.Dial("tcp", s.RedisURL)
-	if err != nil {
-		log.Fatal(err)
-	}
-	s.Redis = conn
-	_, err = s.Redis.Do("SELECT", "1")
-	if err != nil {
-		log.Fatal(err)
-	}
-}
-
-func (s *DatabaseSuite) TearDownSuite() {
-	err := s.Database.Close()
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	err = s.Redis.Close()
-	if err != nil {
-		log.Fatal(err)
-	}
-}
-
-type databaseSuite struct {
-	DatabaseSuite
-}
-
-func (d *databaseSuite) TestGetChannelFromAuthorization() {
-	_, err := d.Database.Exec(`truncate "sources", "channels", "feeds", "subscriptions","items"`)
-	assert.NoError(d.T(), err, "truncate sources, channels, feeds")
-	row := d.Database.QueryRow(`INSERT INTO "channels" (uid, name, created_at, updated_at) VALUES ('abcdef', 'Channel', now(), now()) RETURNING "id"`)
-	var id int
-	err = row.Scan(&id)
-	assert.NoError(d.T(), err, "insert channel")
-	_, err = d.Database.Exec(`INSERT INTO "sources" (channel_id, auth_code, created_at, updated_at) VALUES ($1, '1234', now(), now())`, id)
-	assert.NoError(d.T(), err, "insert sources")
-
-	// source_id found
-	r := httptest.NewRequest("POST", "/micropub?source_id=1234", nil)
-	_, c, err := getChannelFromAuthorization(r, d.Redis, d.Database)
-	assert.NoError(d.T(), err, "channel from source_id")
-	assert.Equal(d.T(), "abcdef", c, "channel uid found")
-
-	// source_id not found
-	r = httptest.NewRequest("POST", "/micropub?source_id=1111", nil)
-	_, c, err = getChannelFromAuthorization(r, d.Redis, d.Database)
-	assert.Error(d.T(), err, "channel from authorization header")
-	assert.Equal(d.T(), "", c, "channel uid found")
-}
-
-func TestDatabaseSuite(t *testing.T) {
-	if testing.Short() {
-		t.Skip("Skip test for database")
-	}
-
-	databaseURL := os.Getenv("DATABASE_TEST_URL")
-	if databaseURL == "" {
-		databaseURL = "host=database user=postgres password=simple dbname=ekster_testing sslmode=disable"
-	}
-	databaseSuite := &databaseSuite{
-		DatabaseSuite{
-			URL:      databaseURL,
-			RedisURL: "redis:6379",
-		},
-	}
-
-	databaseURL = "postgres://postgres@database/ekster_testing?sslmode=disable&user=postgres&password=simple"
-	err := runMigrations(databaseURL)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	suite.Run(t, databaseSuite)
-}

cmd/eksterd/db/migrations/000007_alter_table_add_feed_tier.down.sql

@@ -1,4 +0,0 @@
-alter table "feeds"
-    drop column "tier",
-    drop column "unmodified",
-    drop column "next_fetch_at";

cmd/eksterd/db/migrations/000007_alter_table_add_feed_tier.up.sql

@@ -1,4 +0,0 @@
-alter table "feeds"
-    add column "tier" int default 0,
-    add column "unmodified" int default 0,
-    add column "next_fetch_at" timestamptz;

cmd/eksterd/db/migrations/000008_create_table_sources.down.sql

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

cmd/eksterd/db/migrations/000008_create_table_sources.up.sql

@@ -1,20 +0,0 @@
-BEGIN;
-CREATE OR REPLACE FUNCTION update_timestamp()
-    RETURNS TRIGGER AS $$
-BEGIN
-    NEW.updated_at = now();
-    RETURN NEW;
-END;
-$$ language 'plpgsql';
-COMMIT;
-
-CREATE TABLE "sources" (
-   "id" int primary key generated always as identity,
-   "channel_id" int not null,
-   "auth_code" varchar(64) not null,
-   "created_at" timestamp DEFAULT current_timestamp,
-   "updated_at" timestamp DEFAULT current_timestamp
-);
-
-CREATE TRIGGER sources_update_timestamp BEFORE INSERT OR UPDATE ON "sources"
-FOR EACH ROW EXECUTE PROCEDURE update_timestamp();

cmd/eksterd/http.go

@@ -29,9 +29,11 @@ import (
 	"log"
 	"net/http"
 	"net/url"
+	"os"
 	"strings"
 	"time"
 
+	"github.com/cristalhq/jwt/v4"
 	"p83.nl/go/ekster/pkg/indieauth"
 	"p83.nl/go/ekster/pkg/microsub"
 	"p83.nl/go/ekster/pkg/util"
@@ -44,7 +46,7 @@ import (
 var templates embed.FS
 
 type mainHandler struct {
-	Backend     *memoryBackend
+	Backend     microsub.Microsub
 	BaseURL     string
 	TemplateDir string
 	pool        *redis.Pool
@@ -111,6 +113,12 @@ type authRequest struct {
 	AccessToken string `redis:"access_token"`
 }
 
+type micropubClaims struct {
+	jwt.RegisteredClaims
+	Channel string
+	Me      string
+}
+
 func newMainHandler(backend *memoryBackend, baseURL, templateDir string, pool *redis.Pool) (*mainHandler, error) {
 	h := &mainHandler{Backend: backend}
 
@@ -164,11 +172,13 @@ func loadSession(sessionVar string, conn redis.Conn) (session, error) {
 	sessionKey := "session:" + sessionVar
 	data, err := redis.Values(conn.Do("HGETALL", sessionKey))
 	if err != nil {
+		log.Println(err)
 		return sess, err
 	}
 
 	err = redis.ScanStruct(data, &sess)
 	if err != nil {
+		log.Println(err)
 		return sess, err
 	}
 
@@ -230,18 +240,19 @@ func verifyAuthCode(code, redirectURI, authEndpoint, clientID string) (bool, *au
 	return false, nil, fmt.Errorf("unknown content-type %q while verifying authorization_code", contentType)
 }
 
-func isLoggedIn(backend *memoryBackend, sess *session) bool {
+func isLoggedIn(sess *session) bool {
 	if !sess.LoggedIn {
 		return false
 	}
 
-	if !backend.AuthEnabled {
+	// FIXME: Do we need this?
-		return true
+	// if !backend.AuthEnabled {
-	}
+	// 	return true
-
+	// }
-	if sess.Me != backend.Me {
+	//
-		return false
+	// if sess.Me != backend.Me {
-	}
+	// 	return false
+	// }
 
 	return true
 }
@@ -251,7 +262,6 @@ func performIndieauthCallback(clientID string, r *http.Request, sess *session) (
 	if state != sess.State {
 		return false, &authResponse{}, fmt.Errorf("mismatched state")
 	}
-
 	code := r.Form.Get("code")
 	return verifyAuthCode(code, sess.RedirectURI, sess.AuthorizationEndpoint, clientID)
 }
@@ -377,7 +387,7 @@ func (h *mainHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 
-			if !isLoggedIn(h.Backend, &sess) {
+			if !isLoggedIn(&sess) {
 				w.WriteHeader(401)
 				fmt.Fprintf(w, "Unauthorized")
 				return
@@ -402,14 +412,14 @@ func (h *mainHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			for _, v := range page.Channels {
 				if v.UID == currentChannel {
 					page.CurrentChannel = v
-					if setting, e := h.Backend.Settings[v.UID]; e {
+					// if setting, e := h.Backend.Settings[v.UID]; e {
-						page.CurrentSetting = setting
+					// 	page.CurrentSetting = setting
-					} else {
+					// } else {
 					page.CurrentSetting = channelSetting{}
-					}
+					// }
-					if page.CurrentSetting.ChannelType == "" {
+					// if page.CurrentSetting.ChannelType == "" {
 					page.CurrentSetting.ChannelType = "postgres-stream"
-					}
+					// }
 					page.ExcludedTypeNames = map[string]string{
 						"repost":   "Reposts",
 						"like":     "Likes",
@@ -448,7 +458,7 @@ func (h *mainHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 
-			if !isLoggedIn(h.Backend, &sess) {
+			if !isLoggedIn(&sess) {
 				w.WriteHeader(401)
 				fmt.Fprintf(w, "Unauthorized")
 				return
@@ -476,7 +486,7 @@ func (h *mainHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 
-			if !isLoggedIn(h.Backend, &sess) {
+			if !isLoggedIn(&sess) {
 				w.WriteHeader(401)
 				fmt.Fprintf(w, "Unauthorized")
 				return
@@ -498,9 +508,39 @@ func (h *mainHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			}
 			return
 		} else if r.URL.Path == "/auth" {
-			// check if we are logged in
+			// check arguments for auth
-			// TODO: if not logged in, make sure we get back here
+			query := r.URL.Query()
+			responseType := query.Get("response_type")
+			if responseType != "code" {
+				http.Error(w, "Unsupported response_type", 400)
+				return
+			}
 
+			redirectURI := query.Get("redirect_uri")
+			if _, err := url.Parse(redirectURI); err != nil {
+				http.Error(w, "Missing redirect_uri", 400)
+				return
+			}
+
+			clientID := query.Get("client_id")
+			if _, err := url.Parse(clientID); err != nil {
+				http.Error(w, "Missing client_id", 400)
+				return
+			}
+
+			me := query.Get("me")
+			if _, err := url.Parse(me); err != nil {
+				http.Error(w, "Missing me", 400)
+				return
+			}
+
+			state := query.Get("state")
+			scope := query.Get("scope")
+			if scope == "" {
+				scope = "create"
+			}
+
+			// Check if the client is logging in
 			sessionVar := getSessionCookie(w, r)
 
 			sess, err := loadSession(sessionVar, conn)
@@ -510,7 +550,7 @@ func (h *mainHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 
-			if !isLoggedIn(h.Backend, &sess) {
+			if !isLoggedIn(&sess) {
 				sess.NextURI = r.URL.String()
 				saveSession(sessionVar, &sess, conn)
 				http.Redirect(w, r, "/", http.StatusFound)
@@ -520,18 +560,6 @@ func (h *mainHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			sess.NextURI = r.URL.String()
 			saveSession(sessionVar, &sess, conn)
 
-			query := r.URL.Query()
-
-			// responseType := query.Get("response_type") // TODO: check response_type
-			me := query.Get("me")
-			clientID := query.Get("client_id")
-			redirectURI := query.Get("redirect_uri")
-			state := query.Get("state")
-			scope := query.Get("scope")
-			if scope == "" {
-				scope = "create"
-			}
-
 			authReq := authRequest{
 				Me:          me,
 				ClientID:    clientID,
@@ -585,6 +613,18 @@ func (h *mainHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 			// redirect to endpoint
 			me := r.Form.Get("url")
+			if !strings.HasPrefix("https://", me) {
+				me = fmt.Sprintf("https://%s", me)
+			}
+			meURL, err := url.Parse(me)
+			if err != nil {
+				http.Error(w, fmt.Sprintf("Not a url: %s", me), http.StatusBadRequest)
+				return
+			}
+			if meURL.Path == "" {
+				meURL.Path = "/"
+			}
+			me = meURL.String()
 
 			endpoints, err := getEndpoints(me)
 			if err != nil {
@@ -695,17 +735,31 @@ func (h *mainHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				fmt.Fprintf(w, "ERROR: %q", err)
 				return
 			}
-			token := util.RandStringBytes(32)
+
-			_, err = conn.Do("HMSET", redis.Args{}.Add("token:"+token).AddFlat(&auth)...)
+			key := []byte(os.Getenv("APP_SECRET"))
+			signer, err := jwt.NewSignerHS(jwt.HS256, key)
 			if err != nil {
-				log.Println(err)
+				log.Printf("could not create signer for jwt: %s", err)
-				fmt.Fprintf(w, "ERROR: %q", err)
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+				return
+			}
+			builder := jwt.NewBuilder(signer)
+			token, err := builder.Build(&micropubClaims{
+				RegisteredClaims: jwt.RegisteredClaims{
+					IssuedAt: jwt.NewNumericDate(time.Now()),
+				},
+				Channel: auth.Channel,
+				Me:      auth.Me,
+			})
+			if err != nil {
+				log.Printf("could not create signer for jwt: %s", err)
+				http.Error(w, err.Error(), http.StatusInternalServerError)
 				return
 			}
 
 			res := authTokenResponse{
 				Me:          auth.Me,
-				AccessToken: token,
+				AccessToken: token.String(),
 				TokenType:   "Bearer",
 				Scope:       auth.Scope,
 			}

cmd/eksterd/http_test.go

@@ -0,0 +1,95 @@
+/*
+ *  Ekster is a microsub server
+ *  Copyright (c) 2021 The Ekster authors
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package main
+
+import (
+	"io/ioutil"
+	"math/rand"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+
+	"github.com/gomodule/redigo/redis"
+	"github.com/rafaeljusto/redigomock/v3"
+	"github.com/stretchr/testify/assert"
+	"p83.nl/go/ekster/pkg/server"
+	"p83.nl/go/ekster/pkg/util"
+)
+
+func init() {
+	rand.Seed(1)
+}
+
+func TestMainHandler_ServeHTTP_NoArgs(t *testing.T) {
+	conn := redigomock.NewConn()
+	pool := &redis.Pool{
+		// Return the same connection mock for each Get() call.
+		Dial:    func() (redis.Conn, error) { return conn, nil },
+		MaxIdle: 10,
+	}
+
+	h := mainHandler{Backend: &memoryBackend{AuthEnabled: true}, BaseURL: "", TemplateDir: "", pool: pool}
+	r := httptest.NewRequest(http.MethodGet, "/auth", nil)
+	w := httptest.NewRecorder()
+	h.ServeHTTP(w, r)
+	assert.Equal(t, 400, w.Code)
+}
+func TestMainHandler_ServeHTTP_Args(t *testing.T) {
+	conn := redigomock.NewConn()
+	pool := &redis.Pool{
+		// Return the same connection mock for each Get() call.
+		Dial:    func() (redis.Conn, error) { return conn, nil },
+		MaxIdle: 10,
+	}
+
+	q := url.Values{}
+	q.Add("response_type", "code")
+	q.Add("client_id", "https://example.com/")
+	q.Add("redirect_uri", "https://example.com/callback")
+	q.Add("me", "https://p83.nl/")
+	state := util.RandStringBytes(32)
+	q.Add("state", state)
+	q.Add("scope", "create")
+	h := mainHandler{Backend: &server.NullBackend{}, BaseURL: "", TemplateDir: "", pool: pool}
+	r := httptest.NewRequest(http.MethodGet, "/auth?"+q.Encode(), nil)
+	w := httptest.NewRecorder()
+
+	conn.Command("HGETALL", "session:FpLSjFbcXoEFfRsW").ExpectMap(map[string]string{
+		"logged_in": "1",
+	})
+	conn.Command(
+		"HMSET", "state:XVlBzgbaiCMRAjWwhTHctcuAxhxKQFDa",
+		"me", "https://p83.nl/",
+		"client_id", "https://example.com/",
+		"scope", "create",
+		"redirect_uri", "https://example.com/callback",
+		"state", "XVlBzgbaiCMRAjWwhTHctcuAxhxKQFDa",
+		"code", "",
+		"channel", "",
+		"access_token", "",
+	)
+
+	h.ServeHTTP(w, r)
+	assert.Equal(t, 302, w.Code)
+	body, err := ioutil.ReadAll(w.Result().Body)
+	if assert.NoError(t, err) {
+		assert.Equal(t, "", string(body))
+	}
+}

cmd/eksterd/hubbackend.go

@@ -135,14 +135,15 @@ func (h *hubIncomingBackend) UpdateFeed(processor ContentProcessor, subscription
 	log.Println("UpdateFeed", subscriptionID)
 
 	db := h.database
+	var (
+		topic   string
+		channel string
+		feedID  string
+	)
+
 	// Process all channels that contains this feed
-	rows, err := db.Query(`
+	rows, err := db.Query(
-select topic, c.uid, f.id, c.name
+		`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`,
-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 {
@@ -150,18 +151,16 @@ where s.id = $1
 	}
 
 	for rows.Next() {
-		var topic, channel, feedID, channelName string
+		err = rows.Scan(&topic, &channel, &feedID)
-
-		err = rows.Scan(&topic, &channel, &feedID, &channelName)
 		if err != nil {
 			log.Println(err)
 			continue
 		}
 
-		log.Printf("Updating feed %s %q in %q (%s)\n", feedID, topic, channelName, channel)
+		log.Printf("Updating feed %s %q in %q\n", feedID, topic, channel)
-		_, err = processor.ProcessContent(channel, feedID, topic, contentType, body)
+		err = processor.ProcessContent(channel, feedID, topic, contentType, body)
 		if err != nil {
-			log.Printf("could not process content for channel %s: %s", channelName, err)
+			log.Printf("could not process content for channel %s: %s", channel, err)
 		}
 	}
 
@@ -224,24 +223,22 @@ func (h *hubIncomingBackend) Subscribe(feed *Feed) error {
 }
 
 func (h *hubIncomingBackend) run() error {
-	ticker := time.NewTicker(1 * time.Minute)
+	ticker := time.NewTicker(10 * time.Minute)
 	quit := make(chan struct{})
 
 	go func() {
 		for {
 			select {
 			case <-ticker.C:
-				log.Println("Getting feeds for WebSub started")
+				log.Println("Getting feeds for WebSub")
 				varWebsub.Add("runs", 1)
 
 				feeds, err := h.Feeds()
 				if err != nil {
 					log.Println("Feeds failed:", err)
-					log.Println("Getting feeds for WebSub completed")
 					continue
 				}
 
-				log.Printf("Found %d feeds", len(feeds))
 				for _, feed := range feeds {
 					log.Printf("Looking at %s\n", feed.URL)
 					if feed.ResubscribeAt != nil && time.Now().After(*feed.ResubscribeAt) {
@@ -257,8 +254,6 @@ func (h *hubIncomingBackend) run() error {
 						}
 					}
 				}
-
-				log.Println("Getting feeds for WebSub completed")
 			case <-quit:
 				ticker.Stop()
 				return

cmd/eksterd/main.go

@@ -167,8 +167,8 @@ func main() {
 	// }
 
 	// TODO(peter): automatically gather this information from login or otherwise
-	databaseURL := "postgres://postgres@database/ekster?sslmode=disable&user=postgres&password=simple"
+
-	err := runMigrations(databaseURL)
+	err := runMigrations()
 	if err != nil {
 		log.Fatalf("Error with migrations: %s", err)
 	}
@@ -205,12 +205,12 @@ func (l Log) Verbose() bool {
 	return false
 }
 
-func runMigrations(databaseURL string) error {
+func runMigrations() error {
 	d, err := iofs.New(migrations, "db/migrations")
 	if err != nil {
 		return err
 	}
-	m, err := migrate.NewWithSourceInstance("iofs", d, databaseURL)
+	m, err := migrate.NewWithSourceInstance("iofs", d, "postgres://postgres@database/ekster?sslmode=disable&user=postgres&password=simple")
 	if err != nil {
 		return err
 	}

cmd/eksterd/memory.go

@@ -27,17 +27,14 @@ import (
 	"io"
 	"io/ioutil"
 	"log"
-	"math"
 	"net/http"
 	"net/url"
 	"regexp"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/lib/pq"
-	"github.com/pkg/errors"
 	"p83.nl/go/ekster/pkg/auth"
 	"p83.nl/go/ekster/pkg/fetch"
 	"p83.nl/go/ekster/pkg/microsub"
@@ -105,15 +102,6 @@ type newItemMessage struct {
 	Channel string        `json:"channel"`
 }
 
-type feed struct {
-	UID         string // channel
-	ID          int
-	URL         string
-	Tier        int
-	Unmodified  int
-	NextFetchAt time.Time
-}
-
 func (b *memoryBackend) AuthTokenAccepted(header string, r *auth.TokenResponse) (bool, error) {
 	conn := b.pool.Get()
 	defer func() {
@@ -156,9 +144,6 @@ GROUP BY c.id;
 		}})
 	}
 
-	util.StablePartition(channels, 0, len(channels), func(i int) bool {
-		return channels[i].Unread.HasUnread()
-	})
 	return channels, nil
 }
 
@@ -190,7 +175,7 @@ func (b *memoryBackend) ChannelsCreate(name string) (microsub.Channel, error) {
 	for {
 		varMicrosub.Add("ChannelsCreate.RandStringBytes", 1)
 		channel.UID = util.RandStringBytes(24)
-		result, err := b.database.Exec(`insert into "channels" ("uid", "name", "created_at") values ($1, $2, DEFAULT)`, channel.UID, channel.Name)
+		result, err := b.database.Exec(`insert into "channels" ("uid", "name", "created_at") values($1, $2, DEFAULT)`, channel.UID, channel.Name)
 		if err != nil {
 			log.Println("channels insert", err)
 			if !shouldRetryWithNewUID(err, try) {
@@ -199,7 +184,7 @@ func (b *memoryBackend) ChannelsCreate(name string) (microsub.Channel, error) {
 			try++
 			continue
 		}
-		if n, err := result.RowsAffected(); err == nil {
+		if n, err := result.RowsAffected(); err != nil {
 			if n > 0 {
 				b.broker.Notifier <- sse.Message{Event: "new channel", Object: channelMessage{1, channel}}
 			}
@@ -234,22 +219,15 @@ func (b *memoryBackend) ChannelsDelete(uid string) error {
 	b.broker.Notifier <- sse.Message{Event: "delete channel", Object: channelDeletedMessage{1, uid}}
 	return nil
 }
-func (b *memoryBackend) updateFeed(feed feed) error {
+
-	_, err := b.database.Exec(`
+type feed struct {
-UPDATE "feeds"
+	UID string // channel
-SET "tier" = $2, "unmodified" = $3, "next_fetch_at" = $4
+	ID  int
-WHERE "id" = $1
+	URL string
-`, feed.ID, feed.Tier, feed.Unmodified, feed.NextFetchAt)
-	return err
 }
 
 func (b *memoryBackend) getFeeds() ([]feed, error) {
-	rows, err := b.database.Query(`
+	rows, err := b.database.Query(`SELECT "f"."id", "f"."url", "c"."uid" FROM "feeds" AS "f" INNER JOIN public.channels c on c.id = f.channel_id`)
-SELECT "f"."id", "f"."url", "c"."uid", "f"."tier","f"."unmodified","f"."next_fetch_at"
-FROM "feeds" AS "f"
-INNER JOIN public.channels c ON c.id = f.channel_id
-WHERE next_fetch_at IS NULL OR next_fetch_at < now()
-`)
 	if err != nil {
 		return nil, err
 	}
@@ -258,49 +236,29 @@ WHERE next_fetch_at IS NULL OR next_fetch_at < now()
 	for rows.Next() {
 		var feedID int
 		var feedURL, UID string
-		var tier, unmodified int
-		var nextFetchAt sql.NullTime
 
-		err = rows.Scan(&feedID, &feedURL, &UID, &tier, &unmodified, &nextFetchAt)
+		err = rows.Scan(&feedID, &feedURL, &UID)
 		if err != nil {
 			log.Printf("while scanning feeds: %s", err)
 			continue
 		}
 
-		var fetchTime time.Time
+		feeds = append(feeds, feed{UID, feedID, feedURL})
-		if nextFetchAt.Valid {
-			fetchTime = nextFetchAt.Time
-		} else {
-			fetchTime = time.Now()
-		}
-
-		feeds = append(
-			feeds,
-			feed{
-				UID:         UID,
-				ID:          feedID,
-				URL:         feedURL,
-				Tier:        tier,
-				Unmodified:  unmodified,
-				NextFetchAt: fetchTime,
-			},
-		)
 	}
 
 	return feeds, nil
 }
 
 func (b *memoryBackend) run() {
-	b.ticker = time.NewTicker(1 * time.Minute)
+	b.ticker = time.NewTicker(10 * time.Minute)
 	b.quit = make(chan struct{})
 
 	go func() {
-		b.RefreshFeeds()
-
 		for {
 			select {
 			case <-b.ticker.C:
 				b.RefreshFeeds()
+
 			case <-b.quit:
 				b.ticker.Stop()
 				return
@@ -310,89 +268,43 @@ func (b *memoryBackend) run() {
 }
 
 func (b *memoryBackend) RefreshFeeds() {
-	log.Println("Feed update process started")
-	defer log.Println("Feed update process completed")
-
 	feeds, err := b.getFeeds()
 	if err != nil {
 		return
 	}
 
-	log.Printf("Found %d feeds", len(feeds))
-
 	count := 0
+
 	for _, feed := range feeds {
-		log.Println("Processing", feed.URL)
+		feedURL := feed.URL
-		err := b.refreshFeed(feed)
+		feedID := feed.ID
+		uid := feed.UID
+		log.Println("Processing", feedURL)
+		resp, err := b.Fetch3(uid, feedURL)
 		if err != nil {
-			b.addNotification("Error while fetching feed", feed, err)
+			log.Printf("Error while Fetch3 of %s: %v\n", feedURL, err)
+			b.addNotification("Error while fetching feed", feedURL, err)
+			count++
 			continue
 		}
-
+		err = b.ProcessContent(uid, fmt.Sprintf("%d", feedID), feedURL, resp.Header.Get("Content-Type"), resp.Body)
+		if err != nil {
+			log.Printf("Error while processing content for %s: %v\n", feedURL, err)
+			b.addNotification("Error while processing feed", feedURL, err)
 			count++
+			continue
+		}
+		_ = resp.Body.Close()
 	}
 
 	if count > 0 {
 		_ = b.updateChannelUnreadCount("notifications")
 	}
-	log.Printf("Processed %d feeds", count)
 }
 
-func (b *memoryBackend) refreshFeed(feed feed) error {
+func (b *memoryBackend) addNotification(name string, feedURL string, err error) {
-	resp, err := b.Fetch3(feed.UID, feed.URL)
+	err = b.channelAddItem("notifications", microsub.Item{
-	if err != nil {
-		return fmt.Errorf("while Fetch3 of %s: %w", feed.URL, err)
-	}
-	defer resp.Body.Close()
-
-	changed, err := b.ProcessContent(feed.UID, fmt.Sprintf("%d", feed.ID), feed.URL, resp.Header.Get("Content-Type"), resp.Body)
-	if err != nil {
-		return fmt.Errorf("in ProcessContent of %s: %w", feed.URL, err)
-	}
-
-	if changed {
-		feed.Tier--
-	} else {
-		feed.Unmodified++
-	}
-
-	if feed.Unmodified >= 2 {
-		feed.Tier++
-		feed.Unmodified = 0
-	}
-
-	if feed.Tier > 10 {
-		feed.Tier = 10
-	}
-
-	if feed.Tier < 0 {
-		feed.Tier = 0
-	}
-
-	minutes := time.Duration(math.Ceil(math.Exp2(float64(feed.Tier))))
-
-	feed.NextFetchAt = time.Now().Add(minutes * time.Minute)
-
-	log.Printf("Next Fetch in %d minutes at %v", minutes, feed.NextFetchAt.Format(time.RFC3339))
-
-	err = b.updateFeed(feed)
-	if err != nil {
-		log.Printf("Error: while updating feed %v: %v", feed, err)
-		// don't return error, because it becomes a notification
-		return nil
-	}
-
-	return nil
-}
-
-func (b *memoryBackend) addNotification(name string, feed feed, err error) {
-	_, err = b.channelAddItem("notifications", microsub.Item{
 		Type: "entry",
-		Source: &microsub.Source{
-			ID:   strconv.Itoa(feed.ID),
-			URL:  feed.URL,
-			Name: feed.URL,
-		},
 		Name: name,
 		Content: &microsub.Content{
 			Text: fmt.Sprintf("ERROR: while updating feed: %s", err),
@@ -413,12 +325,9 @@ func (b *memoryBackend) TimelineGet(before, after, channel string) (microsub.Tim
 		return microsub.Timeline{Items: []microsub.Item{}}, err
 	}
 
-	timelineBackend, err := b.getTimeline(channel)
+	timelineBackend := b.getTimeline(channel)
-	if err != nil {
-		return microsub.Timeline{}, err
-	}
 
-	// _ = b.updateChannelUnreadCount(channel)
+	_ = b.updateChannelUnreadCount(channel)
 
 	return timelineBackend.Items(before, after)
 }
@@ -445,7 +354,7 @@ func (b *memoryBackend) FollowGetList(uid string) ([]microsub.Feed, error) {
 }
 
 func (b *memoryBackend) FollowURL(uid string, url string) (microsub.Feed, error) {
-	subFeed := microsub.Feed{Type: "feed", URL: url}
+	feed := microsub.Feed{Type: "feed", URL: url}
 
 	var channelID int
 	err := b.database.QueryRow(`SELECT "id" FROM "channels" WHERE "uid" = $1`, uid).Scan(&channelID)
@@ -458,36 +367,28 @@ func (b *memoryBackend) FollowURL(uid string, url string) (microsub.Feed, error)
 
 	var feedID int
 	err = b.database.QueryRow(
-		`INSERT INTO "feeds" ("channel_id", "url", "tier", "unmodified", "next_fetch_at") VALUES ($1, $2, 1, 0, now()) RETURNING "id"`,
+		`INSERT INTO "feeds" ("channel_id", "url") VALUES ($1, $2) RETURNING "id"`,
 		channelID,
-		subFeed.URL,
+		feed.URL,
 	).Scan(&feedID)
 	if err != nil {
-		return subFeed, err
+		return feed, err
 	}
 
-	var newFeed = feed{
+	resp, err := b.Fetch3(uid, feed.URL)
-		ID:          feedID,
-		UID:         uid,
-		URL:         url,
-		Tier:        1,
-		Unmodified:  0,
-		NextFetchAt: time.Now(),
-	}
-	resp, err := b.Fetch3(uid, subFeed.URL)
 	if err != nil {
 		log.Println(err)
-		b.addNotification("Error while fetching feed", newFeed, err)
+		b.addNotification("Error while fetching feed", feed.URL, err)
 		_ = b.updateChannelUnreadCount("notifications")
-		return subFeed, err
+		return feed, err
 	}
 	defer resp.Body.Close()
 
-	_, _ = b.ProcessContent(uid, fmt.Sprintf("%d", feedID), subFeed.URL, resp.Header.Get("Content-Type"), resp.Body)
+	_ = b.ProcessContent(uid, fmt.Sprintf("%d", feedID), feed.URL, resp.Header.Get("Content-Type"), resp.Body)
 
 	_, _ = b.hubBackend.CreateFeed(url)
 
-	return subFeed, nil
+	return feed, nil
 }
 
 func (b *memoryBackend) UnfollowURL(uid string, url string) error {
@@ -624,16 +525,15 @@ func (b *memoryBackend) PreviewURL(previewURL string) (microsub.Timeline, error)
 }
 
 func (b *memoryBackend) MarkRead(channel string, uids []string) error {
-	tl, err := b.getTimeline(channel)
+	tl := b.getTimeline(channel)
+	err := tl.MarkRead(uids)
+
 	if err != nil {
 		return err
 	}
 
-	if err = tl.MarkRead(uids); err != nil {
+	err = b.updateChannelUnreadCount(channel)
-		return err
+	if err != nil {
-	}
-
-	if err = b.updateChannelUnreadCount(channel); err != nil {
 		return err
 	}
 
@@ -684,35 +584,31 @@ func ProcessSourcedItems(fetcher fetch.Fetcher, fetchURL, contentType string, bo
 
 // ContentProcessor processes content for a channel and feed
 type ContentProcessor interface {
-	ProcessContent(channel, feedID, fetchURL, contentType string, body io.Reader) (bool, error)
+	ProcessContent(channel, feedID, fetchURL, contentType string, body io.Reader) error
 }
 
-// ProcessContent processes content of a feed, returns if the feed has changed or not
+func (b *memoryBackend) ProcessContent(channel, feedID, fetchURL, contentType string, body io.Reader) error {
-func (b *memoryBackend) ProcessContent(channel, feedID, fetchURL, contentType string, body io.Reader) (bool, error) {
 	cachingFetch := WithCaching(b.pool, fetch.FetcherFunc(Fetch2))
 
 	items, err := ProcessSourcedItems(cachingFetch, fetchURL, contentType, body)
 	if err != nil {
-		return false, err
+		return err
 	}
 
-	changed := false
-
 	for _, item := range items {
 		item.Source.ID = feedID
-		added, err := b.channelAddItemWithMatcher(channel, item)
+		err = b.channelAddItemWithMatcher(channel, item)
 		if err != nil {
 			log.Printf("ERROR: (feedID=%s) %s\n", feedID, err)
 		}
-		changed = changed || added
 	}
 
 	err = b.updateChannelUnreadCount(channel)
 	if err != nil {
-		return changed, err
+		return err
 	}
 
-	return changed, nil
+	return nil
 }
 
 // Fetch3 fills stuff
@@ -721,12 +617,17 @@ func (b *memoryBackend) Fetch3(channel, fetchURL string) (*http.Response, error)
 	return Fetch2(fetchURL)
 }
 
-func (b *memoryBackend) channelAddItemWithMatcher(channel string, item microsub.Item) (bool, error) {
+func (b *memoryBackend) channelAddItemWithMatcher(channel string, item microsub.Item) error {
 	// an item is posted
 	// check for all channels as channel
 	// if regex matches item
 	//  - add item to channel
 
+	err := addToSearch(item, channel)
+	if err != nil {
+		return fmt.Errorf("addToSearch in channelAddItemWithMatcher: %v", err)
+	}
+
 	var updatedChannels []string
 
 	b.lock.RLock()
@@ -739,23 +640,23 @@ func (b *memoryBackend) channelAddItemWithMatcher(channel string, item microsub.
 				switch v {
 				case "repost":
 					if len(item.RepostOf) > 0 {
-						return false, nil
+						return nil
 					}
 				case "like":
 					if len(item.LikeOf) > 0 {
-						return false, nil
+						return nil
 					}
 				case "bookmark":
 					if len(item.BookmarkOf) > 0 {
-						return false, nil
+						return nil
 					}
 				case "reply":
 					if len(item.InReplyTo) > 0 {
-						return false, nil
+						return nil
 					}
 				case "checkin":
 					if item.Checkin != nil {
-						return false, nil
+						return nil
 					}
 				}
 			}
@@ -764,27 +665,19 @@ func (b *memoryBackend) channelAddItemWithMatcher(channel string, item microsub.
 			re, err := regexp.Compile(setting.IncludeRegex)
 			if err != nil {
 				log.Printf("error in regexp: %q, %s\n", setting.IncludeRegex, err)
-				return false, nil
+				return nil
 			}
 
 			if matchItem(item, re) {
 				log.Printf("Included %#v\n", item)
-				added, err := b.channelAddItem(channelKey, item)
+				err := b.channelAddItem(channelKey, item)
 				if err != nil {
 					continue
 				}
-
-				err = addToSearch(item, channel)
-				if err != nil {
-					return added, fmt.Errorf("addToSearch in channelAddItemWithMatcher: %v", err)
-				}
-
-				if added {
 				updatedChannels = append(updatedChannels, channelKey)
 			}
 		}
 	}
-	}
 
 	// Update all channels that have added items, because of the include matching
 	for _, value := range updatedChannels {
@@ -804,26 +697,15 @@ func (b *memoryBackend) channelAddItemWithMatcher(channel string, item microsub.
 		excludeRegex, err := regexp.Compile(setting.ExcludeRegex)
 		if err != nil {
 			log.Printf("error in regexp: %q\n", excludeRegex)
-			return false, nil
+			return nil
 		}
 		if matchItem(item, excludeRegex) {
 			log.Printf("Excluded %#v\n", item)
-			return false, nil
+			return nil
 		}
 	}
 
-	added, err := b.channelAddItem(channel, item)
+	return b.channelAddItem(channel, item)
-
-	if err != nil {
-		return added, err
-	}
-
-	err = addToSearch(item, channel)
-	if err != nil {
-		return added, fmt.Errorf("addToSearch in channelAddItemWithMatcher: %v", err)
-	}
-
-	return added, nil
 }
 
 func matchItem(item microsub.Item, re *regexp.Regexp) bool {
@@ -852,15 +734,11 @@ func matchItemText(item microsub.Item, re *regexp.Regexp) bool {
 	return re.MatchString(item.Name)
 }
 
-func (b *memoryBackend) channelAddItem(channel string, item microsub.Item) (bool, error) {
+func (b *memoryBackend) channelAddItem(channel string, item microsub.Item) error {
-	timelineBackend, err := b.getTimeline(channel)
+	timelineBackend := b.getTimeline(channel)
-	if err != nil {
-		return false, err
-	}
-
 	added, err := timelineBackend.AddItem(item)
 	if err != nil {
-		return added, err
+		return err
 	}
 
 	// Sent message to Server-Sent-Events
@@ -868,33 +746,23 @@ func (b *memoryBackend) channelAddItem(channel string, item microsub.Item) (bool
 		b.broker.Notifier <- sse.Message{Event: "new item", Object: newItemMessage{item, channel}}
 	}
 
-	return added, err
+	return err
 }
 
-// ErrNotUpdated is used when the unread count is not updated
-var ErrNotUpdated = errors.New("timeline unread count not updated")
-
-// ErrNotFound is used when the timeline is not found
-var ErrNotFound = errors.New("timeline not found")
-
 func (b *memoryBackend) updateChannelUnreadCount(channel string) error {
-	tl, err := b.getTimeline(channel)
+	// tl := b.getTimeline(channel)
-	if err != nil {
+	// unread, err := tl.Count()
-		return err
+	// if err != nil {
-	}
+	// 	return err
-
+	// }
-	unread, err := tl.Count()
+	//
-	if err != nil {
+	// currentCount := c.Unread.UnreadCount
-		return ErrNotUpdated
+	// c.Unread = microsub.Unread{Type: microsub.UnreadCount, UnreadCount: unread}
-	}
+	//
-
+	// // Sent message to Server-Sent-Events
-	var c = microsub.Channel{
+	// if currentCount != unread {
-		UID:    channel,
+	// 	b.broker.Notifier <- sse.Message{Event: "new item in channel", Object: c}
-		Unread: microsub.Unread{Type: microsub.UnreadCount, UnreadCount: unread},
+	// }
-	}
-
-	// Sent message to Server-Sent-Events
-	b.broker.Notifier <- sse.Message{Event: "new item in channel", Object: c}
 
 	return nil
 }
@@ -974,12 +842,15 @@ func Fetch2(fetchURL string) (*http.Response, error) {
 	return resp, err
 }
 
-func (b *memoryBackend) getTimeline(channel string) (timeline.Backend, error) {
+func (b *memoryBackend) getTimeline(channel string) timeline.Backend {
 	// Set a default timeline type if not set
 	timelineType := "postgres-stream"
+	// if setting, ok := b.Settings[channel]; ok && setting.ChannelType != "" {
+	// 	timelineType = setting.ChannelType
+	// }
 	tl := timeline.Create(channel, timelineType, b.pool, b.database)
 	if tl == nil {
-		return tl, fmt.Errorf("timeline id %q: %w", channel, ErrNotFound)
+		log.Printf("no timeline found with name %q and type %q", channel, timelineType)
 	}
-	return tl, nil
+	return tl
 }

cmd/eksterd/micropub.go

@@ -20,11 +20,11 @@ package main
 
 import (
 	"crypto/sha1"
-	"database/sql"
 	"encoding/json"
 	"fmt"
 	"log"
 	"net/http"
+	"os"
 	"strings"
 	"time"
 
@@ -34,6 +34,8 @@ import (
 	"github.com/gomodule/redigo/redis"
 	"github.com/pkg/errors"
 	"willnorris.com/go/microformats"
+
+	"github.com/cristalhq/jwt/v4"
 )
 
 type micropubHandler struct {
@@ -66,7 +68,7 @@ func (h *micropubHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	if r.Method == http.MethodPost {
 		var channel string
 
-		sourceID, channel, err := getChannelFromAuthorization(r, conn, h.Backend.database)
+		channel, err = getChannelFromAuthorization(r, conn)
 		if err != nil {
 			log.Println(err)
 			http.Error(w, "unauthorized", http.StatusUnauthorized)
@@ -100,12 +102,7 @@ func (h *micropubHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 		item.ID = newID
 
-		item.Source = &microsub.Source{
+		err = h.Backend.channelAddItemWithMatcher(channel, *item)
-			ID:   fmt.Sprintf("micropub:%d", sourceID),
-			Name: fmt.Sprintf("Source %d", sourceID),
-		}
-
-		_, err = h.Backend.channelAddItemWithMatcher(channel, *item)
 		if err != nil {
 			log.Printf("could not add item to channel %s: %v", channel, err)
 		}
@@ -171,23 +168,16 @@ func parseIncomingItem(r *http.Request) (*microsub.Item, error) {
 	return nil, fmt.Errorf("content-type %q is not supported", contentType)
 }
 
-func getChannelFromAuthorization(r *http.Request, conn redis.Conn, database *sql.DB) (int, string, error) {
+func getChannelFromAuthorization(r *http.Request, conn redis.Conn) (string, error) {
 	// backward compatible
 	sourceID := r.URL.Query().Get("source_id")
 	if sourceID != "" {
-		row := database.QueryRow(`
+		channel, err := redis.String(conn.Do("HGET", "sources", sourceID))
-SELECT s.id as source_id, c.uid
+		if err != nil {
-FROM "sources" AS "s"
+			return "", errors.Wrapf(err, "could not get channel for sourceID: %s", sourceID)
-INNER JOIN "channels" AS "c" ON s.channel_id = c.id
-WHERE "auth_code" = $1
-`, sourceID)
-
-		var channel string
-		var sourceID int
-		if err := row.Scan(&sourceID, &channel); err == sql.ErrNoRows {
-			return 0, "", errors.New("channel not found")
 		}
-		return sourceID, channel, nil
+
+		return channel, nil
 	}
 
 	// full micropub with indieauth
@@ -196,11 +186,41 @@ WHERE "auth_code" = $1
 		token := authHeader[7:]
 		channel, err := redis.String(conn.Do("HGET", "token:"+token, "channel"))
 		if err != nil {
-			return 0, "", errors.Wrap(err, "could not get channel for token")
+			_, err := verifyJWT(token)
+			if err != nil {
+				return "", err
 			}
 
-		return 0, channel, nil
+			return "", errors.Wrap(err, "could not get channel for token")
 		}
 
-	return 0, "", fmt.Errorf("could not get channel from authorization")
+		return channel, nil
+	}
+
+	return "", fmt.Errorf("could not get channel from authorization")
+}
+
+func verifyJWT(token string) (bool, error) {
+	key := []byte(os.Getenv("APP_SECRET"))
+	verifier, err := jwt.NewVerifierHS(jwt.HS256, key)
+	if err != nil {
+		return false, fmt.Errorf("could not create verifier for jwt: %w", err)
+	}
+
+	newToken, err := jwt.Parse([]byte(token), verifier)
+	if err != nil {
+		return false, fmt.Errorf("could not parse jwt: %w", err)
+	}
+
+	var newClaims jwt.RegisteredClaims
+	err = json.Unmarshal(newToken.Claims(), &newClaims)
+	if err != nil {
+		return false, fmt.Errorf("could not parse jwt claims: %w", err)
+	}
+
+	if !newClaims.IsValidAt(time.Now()) {
+		return false, fmt.Errorf("jwt is not valid now")
+	}
+
+	return true, nil
 }

go.mod

@@ -5,14 +5,15 @@ go 1.16
 require (
 	github.com/axgle/mahonia v0.0.0-20180208002826-3358181d7394
 	github.com/blevesearch/bleve/v2 v2.0.3
+	github.com/cristalhq/jwt/v4 v4.0.0-beta // indirect
 	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
+	github.com/gomodule/redigo v1.8.5
 	github.com/lib/pq v1.10.1
 	github.com/pkg/errors v0.9.1
+	github.com/rafaeljusto/redigomock/v3 v3.0.1 // indirect
 	github.com/stretchr/testify v1.7.0
 	golang.org/x/net v0.0.0-20211013171255-e13a2654a71e
-	golang.org/x/text v0.3.7
 	willnorris.com/go/microformats v1.1.0
 )

go.sum

@@ -312,6 +312,8 @@ github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:ma
 github.com/cpuguy83/go-md2man/v2 v2.0.0/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
 github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/cristalhq/jwt/v4 v4.0.0-beta h1:TSNUSW7CCvMtX+Rg3vj706BwqIEt9HMvVv6Syifq5mU=
+github.com/cristalhq/jwt/v4 v4.0.0-beta/go.mod h1:HnYraSNKDRag1DZP92rYHyrjyQHnVEHPNqesmzs+miQ=
 github.com/cyphar/filepath-securejoin v0.2.2/go.mod h1:FpkQEhXnPnOthhzymB7CGsFk2G9VLXONKD9G7QGMM+4=
 github.com/cznic/mathutil v0.0.0-20180504122225-ca4c9f2c1369/go.mod h1:e6NPNENfs9mPDVNRekM7lKScauxd5kXTr1Mfyig6TDM=
 github.com/d2g/dhcp4 v0.0.0-20170904100407-a1d1b6c41b1c/go.mod h1:Ct2BUK8SB0YC1SMSibvLzxjeJLnrYEVLULFNiHY9YfQ=
@@ -496,6 +498,9 @@ github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
 github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/gomodule/redigo v1.8.2 h1:H5XSIre1MB5NbPYFp+i1NBbb5qN1W8Y8YAQoAYbkm8k=
 github.com/gomodule/redigo v1.8.2/go.mod h1:P9dn9mFrCBvWhGE1wpxx6fgq7BAeLBk+UUUzlpkBYO0=
+github.com/gomodule/redigo v1.8.3/go.mod h1:P9dn9mFrCBvWhGE1wpxx6fgq7BAeLBk+UUUzlpkBYO0=
+github.com/gomodule/redigo v1.8.5 h1:nRAxCa+SVsyjSBrtZmG/cqb6VbTmuRzpg/PoTFlpumc=
+github.com/gomodule/redigo v1.8.5/go.mod h1:P9dn9mFrCBvWhGE1wpxx6fgq7BAeLBk+UUUzlpkBYO0=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/flatbuffers v2.0.0+incompatible/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8=
@@ -838,6 +843,8 @@ github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4O
 github.com/prometheus/procfs v0.2.0/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
+github.com/rafaeljusto/redigomock/v3 v3.0.1 h1:AUsXTuf+UEMwVEgRHRDYFFCJ1quS2JVDQmTWypjI5mI=
+github.com/rafaeljusto/redigomock/v3 v3.0.1/go.mod h1:51LNR7Q4YFsi0N+CHr7+FC1Jx2lPLzcRHCPlLO2Qbpw=
 github.com/rcrowley/go-metrics v0.0.0-20190826022208-cac0b30c2563/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/remyoudompheng/bigfft v0.0.0-20190728182440-6a916e37a237/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
@@ -1234,7 +1241,6 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=

pkg/microsub/protocol.go

@@ -155,6 +155,7 @@ type Microsub interface {
 	ItemSearch(channel, query string) ([]Item, error)
 
 	Events() (chan sse.Message, error)
+	RefreshFeeds()
 }
 
 // MarshalJSON encodes an Unread value as JSON

pkg/rss/rss_1.0.go

@@ -5,10 +5,8 @@ import (
 	"encoding/xml"
 	"fmt"
 	"sort"
+	"strings"
 	"time"
-
-	"golang.org/x/text/cases"
-	"golang.org/x/text/language"
 )
 
 func parseRSS1(data []byte) (*Feed, error) {
@@ -31,8 +29,6 @@ func parseRSS1(data []byte) (*Feed, error) {
 	out.Description = channel.Description
 	out.Link = channel.Link
 	out.Image = channel.Image.Image()
-
-	titleCaser := cases.Title(language.English)
 	if channel.MinsToLive != 0 {
 		sort.Ints(channel.SkipHours)
 		next := time.Now().Add(time.Duration(channel.MinsToLive) * time.Minute)
@@ -45,7 +41,7 @@ func parseRSS1(data []byte) (*Feed, error) {
 		for trying {
 			trying = false
 			for _, day := range channel.SkipDays {
-				if titleCaser.String(day) == next.Weekday().String() {
+				if strings.Title(day) == next.Weekday().String() {
 					next.Add(time.Duration(24-next.Hour()) * time.Hour)
 					trying = true
 					break

pkg/rss/rss_2.0.go

@@ -5,10 +5,8 @@ import (
 	"encoding/xml"
 	"fmt"
 	"sort"
+	"strings"
 	"time"
-
-	"golang.org/x/text/cases"
-	"golang.org/x/text/language"
 )
 
 func parseRSS2(data []byte) (*Feed, error) {
@@ -40,7 +38,6 @@ func parseRSS2(data []byte) (*Feed, error) {
 
 	out.Image = channel.Image.Image()
 	if channel.MinsToLive != 0 {
-		titleCaser := cases.Title(language.English)
 		sort.Ints(channel.SkipHours)
 		next := time.Now().Add(time.Duration(channel.MinsToLive) * time.Minute)
 		for _, hour := range channel.SkipHours {
@@ -52,7 +49,7 @@ func parseRSS2(data []byte) (*Feed, error) {
 		for trying {
 			trying = false
 			for _, day := range channel.SkipDays {
-				if titleCaser.String(day) == next.Weekday().String() {
+				if strings.Title(day) == next.Weekday().String() {
 					next.Add(time.Duration(24-next.Hour()) * time.Hour)
 					trying = true
 					break

pkg/server/microsub.go

@@ -239,7 +239,6 @@ func (h *microsubHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 					})
 					return
 				}
-				log.Printf("Searching for %s in %s (%d results)", query, channel, len(items))
 				respondJSON(w, map[string]interface{}{
 					"query": query,
 					"items": items,

pkg/server/null.go

@@ -27,6 +27,12 @@ import (
 type NullBackend struct {
 }
 
+// RefreshFeeds refreshes feeds
+func (b *NullBackend) RefreshFeeds() {
+	// TODO implement me
+	// panic("implement me")
+}
+
 // ChannelsGetList gets no channels
 func (b *NullBackend) ChannelsGetList() ([]microsub.Channel, error) {
 	return []microsub.Channel{

pkg/timeline/postgres.go

@@ -53,11 +53,52 @@ func (p *postgresStream) Init() error {
 	if err != nil {
 		return fmt.Errorf("database ping failed: %w", err)
 	}
+	//
+	// 	_, err = conn.ExecContext(ctx, `
+	// CREATE TABLE IF NOT EXISTS "channels" (
+	//     "id" int primary key generated always as identity,
+	//     "name" varchar(255) unique,
+	//     "created_at" timestamp DEFAULT current_timestamp
+	// );
+	// `)
+	// 	if err != nil {
+	// 		return fmt.Errorf("create channels table failed: %w", err)
+	// 	}
+	//
+	// 	_, err = conn.ExecContext(ctx, `
+	// CREATE TABLE IF NOT EXISTS "items" (
+	//     "id" int primary key generated always as identity,
+	//     "channel_id" int references "channels" on delete cascade,
+	//     "uid" varchar(512) not null unique,
+	//     "is_read" int default 0,
+	//     "data" jsonb,
+	//     "created_at" timestamp DEFAULT current_timestamp,
+	//     "updated_at" timestamp,
+	//     "published_at" timestamp
+	// );
+	// `)
+	// 	if err != nil {
+	// 		return fmt.Errorf("create items table failed: %w", err)
+	// 	}
+	//
+	// 	_, err = conn.ExecContext(ctx, `ALTER TABLE "items" ALTER COLUMN "data" TYPE jsonb, ALTER COLUMN "uid"  TYPE varchar(1024)`)
+	// 	if err != nil {
+	// 		return fmt.Errorf("alter items table failed: %w", err)
+	// 	}
+
+	_, err = conn.ExecContext(ctx, `INSERT INTO "channels" ("uid", "name", "created_at") VALUES ($1, $1, DEFAULT)
+ 		ON CONFLICT DO NOTHING`, p.channel)
+	if err != nil {
+		return fmt.Errorf("create channel failed: %w", err)
+	}
 
 	row := conn.QueryRowContext(ctx, `SELECT "id" FROM "channels" WHERE "uid" = $1`, p.channel)
+	if row == nil {
+		return fmt.Errorf("fetch channel failed: %w", err)
+	}
 	err = row.Scan(&p.channelID)
-	if err == sql.ErrNoRows {
+	if err != nil {
-		return fmt.Errorf("channel %s not found: %w", p.channel, err)
+		return fmt.Errorf("fetch channel failed while scanning: %w", err)
 	}
 
 	return nil
@@ -86,16 +127,16 @@ WHERE "channel_id" = $1
 			log.Println(err)
 		} else {
 			args = append(args, b)
-			qb.WriteString(` AND "published_at" > $2`)
+			qb.WriteString(` AND "published_at" < $2`)
 		}
 	} else if after != "" {
 		b, err := time.Parse(time.RFC3339, after)
 		if err == nil {
 			args = append(args, b)
-			qb.WriteString(` AND "published_at" < $2`)
+			qb.WriteString(` AND "published_at" > $2`)
 		}
 	}
-	qb.WriteString(` ORDER BY "published_at" DESC LIMIT 20`)
+	qb.WriteString(` ORDER BY "published_at" DESC LIMIT 10`)
 
 	rows, err := conn.QueryContext(context.Background(), qb.String(), args...)
 	if err != nil {
@@ -139,12 +180,9 @@ WHERE "channel_id" = $1
 		return tl, err
 	}
 
-	if len(tl.Items) > 0 && hasMoreBefore(conn, tl.Items[0].Published) {
+	// TODO: should only be set of there are more items available
-		tl.Paging.Before = tl.Items[0].Published
+	tl.Paging.Before = last
-	}
+	// tl.Paging.After = last
-	if hasMoreAfter(conn, last) {
-		tl.Paging.After = last
-	}
 
 	if tl.Items == nil {
 		tl.Items = []microsub.Item{}
@@ -153,24 +191,6 @@ WHERE "channel_id" = $1
 	return tl, nil
 }
 
-func hasMoreBefore(conn *sql.Conn, before string) bool {
-	row := conn.QueryRowContext(context.Background(), `SELECT COUNT(*) FROM "items" WHERE "published_at" > $1`, before)
-	var count int
-	if err := row.Scan(&count); err == sql.ErrNoRows {
-		return false
-	}
-	return count > 0
-}
-
-func hasMoreAfter(conn *sql.Conn, after string) bool {
-	row := conn.QueryRowContext(context.Background(), `SELECT COUNT(*) FROM "items" WHERE "published_at" < $1`, after)
-	var count int
-	if err := row.Scan(&count); err == sql.ErrNoRows {
-		return false
-	}
-	return count > 0
-}
-
 // Count
 func (p *postgresStream) Count() (int, error) {
 	ctx := context.Background()
@@ -179,12 +199,16 @@ func (p *postgresStream) Count() (int, error) {
 		return -1, err
 	}
 	defer conn.Close()
-	var count int
 	row := conn.QueryRowContext(context.Background(), `SELECT COUNT(*) FROM items WHERE channel_id = $1 AND "is_read" = 0`, p.channelID)
-	err = row.Scan(&count)
+	if row == nil {
-	if err != nil && err == sql.ErrNoRows {
 		return 0, nil
 	}
+	var count int
+	err = row.Scan(&count)
+	if err != nil {
+		return -1, err
+	}
+
 	return count, nil
 }