Compare commits

..

1 Commits

Author SHA1 Message Date
d91b3d72b3
WIP: verify JWT
Some checks failed
continuous-integration/drone/push Build is failing
2022-04-16 13:32:54 +02:00
22 changed files with 452 additions and 558 deletions

View File

@ -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 ./...
---
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
- staticcheck ./...
- 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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -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 {
return true
}
if sess.Me != backend.Me {
return false
}
// FIXME: Do we need this?
// if !backend.AuthEnabled {
// return true
// }
//
// 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 {
page.CurrentSetting = setting
} else {
page.CurrentSetting = channelSetting{}
}
if page.CurrentSetting.ChannelType == "" {
page.CurrentSetting.ChannelType = "postgres-stream"
}
// if setting, e := h.Backend.Settings[v.UID]; e {
// page.CurrentSetting = setting
// } else {
page.CurrentSetting = channelSetting{}
// }
// 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
// TODO: if not logged in, make sure we get back here
// check arguments for auth
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)
fmt.Fprintf(w, "ERROR: %q", err)
log.Printf("could not create signer for jwt: %s", 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,
}

95
cmd/eksterd/http_test.go Normal file
View File

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

View File

@ -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(`
select topic, c.uid, f.id, c.name
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
`,
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 {
@ -150,18 +151,16 @@ where s.id = $1
}
for rows.Next() {
var topic, channel, feedID, channelName string
err = rows.Scan(&topic, &channel, &feedID, &channelName)
err = rows.Scan(&topic, &channel, &feedID)
if err != nil {
log.Println(err)
continue
}
log.Printf("Updating feed %s %q in %q (%s)\n", feedID, topic, channelName, channel)
_, err = processor.ProcessContent(channel, feedID, topic, contentType, body)
log.Printf("Updating feed %s %q in %q\n", feedID, topic, channel)
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

View File

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

View File

@ -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(`
UPDATE "feeds"
SET "tier" = $2, "unmodified" = $3, "next_fetch_at" = $4
WHERE "id" = $1
`, feed.ID, feed.Tier, feed.Unmodified, feed.NextFetchAt)
return err
type feed struct {
UID string // channel
ID int
URL string
}
func (b *memoryBackend) getFeeds() ([]feed, error) {
rows, err := b.database.Query(`
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()
`)
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`)
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
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,
},
)
feeds = append(feeds, feed{UID, feedID, feedURL})
}
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)
err := b.refreshFeed(feed)
feedURL := feed.URL
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
}
count++
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 {
resp, err := b.Fetch3(feed.UID, feed.URL)
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{
func (b *memoryBackend) addNotification(name string, feedURL string, 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)
if err != nil {
return microsub.Timeline{}, err
}
timelineBackend := b.getTimeline(channel)
// _ = 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{
ID: feedID,
UID: uid,
URL: url,
Tier: 1,
Unmodified: 0,
NextFetchAt: time.Now(),
}
resp, err := b.Fetch3(uid, subFeed.URL)
resp, err := b.Fetch3(uid, feed.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 {
return err
}
if err = b.updateChannelUnreadCount(channel); err != nil {
err = b.updateChannelUnreadCount(channel)
if 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) (bool, error) {
func (b *memoryBackend) ProcessContent(channel, feedID, fetchURL, contentType string, body io.Reader) 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,24 +665,16 @@ 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)
}
updatedChannels = append(updatedChannels, channelKey)
}
}
}
@ -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)
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
return b.channelAddItem(channel, item)
}
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) {
timelineBackend, err := b.getTimeline(channel)
if err != nil {
return false, err
}
func (b *memoryBackend) channelAddItem(channel string, item microsub.Item) error {
timelineBackend := b.getTimeline(channel)
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)
if err != nil {
return err
}
unread, err := tl.Count()
if err != nil {
return ErrNotUpdated
}
var c = microsub.Channel{
UID: channel,
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}
// tl := b.getTimeline(channel)
// unread, err := tl.Count()
// if err != nil {
// return err
// }
//
// currentCount := c.Unread.UnreadCount
// c.Unread = microsub.Unread{Type: microsub.UnreadCount, UnreadCount: unread}
//
// // Sent message to Server-Sent-Events
// if currentCount != unread {
// 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
}

View File

@ -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{
ID: fmt.Sprintf("micropub:%d", sourceID),
Name: fmt.Sprintf("Source %d", sourceID),
}
_, err = h.Backend.channelAddItemWithMatcher(channel, *item)
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(`
SELECT s.id as source_id, c.uid
FROM "sources" AS "s"
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")
channel, err := redis.String(conn.Do("HGET", "sources", sourceID))
if err != nil {
return "", errors.Wrapf(err, "could not get channel for sourceID: %s", sourceID)
}
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 "", errors.Wrap(err, "could not get channel for token")
}
return 0, channel, nil
return channel, nil
}
return 0, "", fmt.Errorf("could not get channel from authorization")
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
}

5
go.mod
View File

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

8
go.sum
View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -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 {
return fmt.Errorf("channel %s not found: %w", p.channel, err)
if err != nil {
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) {
tl.Paging.Before = tl.Items[0].Published
}
if hasMoreAfter(conn, last) {
tl.Paging.After = last
}
// TODO: should only be set of there are more items available
tl.Paging.Before = 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 err != nil && err == sql.ErrNoRows {
if row == nil {
return 0, nil
}
var count int
err = row.Scan(&count)
if err != nil {
return -1, err
}
return count, nil
}