mirror of
https://github.com/mxpv/podsync.git
synced 2024-05-11 05:55:04 +00:00
Switch to Postgres storage for feeds
This commit is contained in:
Maksym Pavlenko
@@ -1,76 +0,0 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"log"
|
||||
"net"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/GoogleCloudPlatform/cloudsql-proxy/proxy/proxy"
|
||||
"github.com/go-pg/pg"
|
||||
"github.com/mxpv/podsync/pkg/model"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
type PgConfig struct {
|
||||
ConnectionUrl string `yaml:"connectionUrl"`
|
||||
}
|
||||
|
||||
type PgStorage struct {
|
||||
db *pg.DB
|
||||
}
|
||||
|
||||
func (p *PgStorage) CreateFeed(feed *model.Feed) error {
|
||||
feed.LastAccess = time.Now().UTC()
|
||||
_, err := p.db.Model(feed).OnConflict("DO NOTHING").Insert()
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to create feed")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *PgStorage) GetFeed(hashId string) (*model.Feed, error) {
|
||||
lastAccess := time.Now().UTC()
|
||||
|
||||
feed := &model.Feed{}
|
||||
_, err := p.db.Model(feed).
|
||||
Set("last_access = ?", lastAccess).
|
||||
Where("hash_id = ?", hashId).
|
||||
Returning("*").
|
||||
Update()
|
||||
|
||||
return feed, err
|
||||
}
|
||||
|
||||
func NewPgStorage(config *PgConfig) (*PgStorage, error) {
|
||||
opts, err := pg.ParseURL(config.ConnectionUrl)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// If host format is "projection:region:host", than use Google SQL Proxy
|
||||
// See https://github.com/go-pg/pg/issues/576
|
||||
if strings.Count(opts.Addr, ":") == 2 {
|
||||
log.Print("using GCP SQL proxy")
|
||||
opts.Dialer = func(network, addr string) (net.Conn, error) {
|
||||
return proxy.Dial(addr)
|
||||
}
|
||||
}
|
||||
|
||||
db := pg.Connect(opts)
|
||||
|
||||
// Check database connectivity
|
||||
if _, err := db.ExecOne("SELECT 1"); err != nil {
|
||||
db.Close()
|
||||
return nil, errors.Wrap(err, "failed to check database connectivity")
|
||||
}
|
||||
|
||||
log.Print("running update script")
|
||||
if _, err := db.Exec(installScript); err != nil {
|
||||
return nil, errors.Wrap(err, "failed to upgrade database structure")
|
||||
}
|
||||
|
||||
storage := &PgStorage{db: db}
|
||||
return storage, nil
|
||||
}
|
||||
@@ -1,38 +0,0 @@
|
||||
package storage
|
||||
|
||||
const installScript = `
|
||||
BEGIN;
|
||||
|
||||
DO $$
|
||||
BEGIN
|
||||
IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'provider') THEN
|
||||
CREATE TYPE provider AS ENUM ('youtube', 'vimeo');
|
||||
END IF;
|
||||
IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'link_type') THEN
|
||||
CREATE TYPE link_type AS ENUM ('channel', 'playlist', 'user', 'group');
|
||||
END IF;
|
||||
IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'quality') THEN
|
||||
CREATE TYPE quality AS ENUM ('high', 'low');
|
||||
END IF;
|
||||
IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'format') THEN
|
||||
CREATE TYPE format AS ENUM ('audio', 'video');
|
||||
END IF;
|
||||
END
|
||||
$$;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS feeds (
|
||||
feed_id BIGSERIAL PRIMARY KEY,
|
||||
hash_id VARCHAR(12) NOT NULL CHECK (hash_id <> '') UNIQUE,
|
||||
user_id VARCHAR(32) NULL,
|
||||
item_id VARCHAR(32) NOT NULL CHECK (item_id <> ''),
|
||||
provider provider NOT NULL,
|
||||
link_type link_type NOT NULL,
|
||||
page_size INT NOT NULL DEFAULT 50,
|
||||
format format NOT NULL DEFAULT 'video',
|
||||
quality quality NOT NULL DEFAULT 'high',
|
||||
feature_level INT NOT NULL DEFAULT 0,
|
||||
last_access timestamp WITHOUT TIME ZONE NOT NULL
|
||||
);
|
||||
|
||||
COMMIT;
|
||||
`
|
||||
@@ -1,107 +0,0 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/mxpv/podsync/pkg/api"
|
||||
"github.com/mxpv/podsync/pkg/model"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestPgStorage_CreateFeed(t *testing.T) {
|
||||
feed := &model.Feed{
|
||||
HashID: "xyz",
|
||||
Provider: api.ProviderYoutube,
|
||||
LinkType: api.LinkTypeChannel,
|
||||
ItemID: "123",
|
||||
}
|
||||
|
||||
client := createClient(t)
|
||||
err := client.CreateFeed(feed)
|
||||
require.NoError(t, err)
|
||||
require.True(t, feed.FeedID > 0)
|
||||
}
|
||||
|
||||
func TestPgStorage_CreateFeedWithDuplicate(t *testing.T) {
|
||||
feed := &model.Feed{
|
||||
HashID: "123",
|
||||
Provider: api.ProviderYoutube,
|
||||
LinkType: api.LinkTypeChannel,
|
||||
ItemID: "123",
|
||||
}
|
||||
|
||||
client := createClient(t)
|
||||
err := client.CreateFeed(feed)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Ensure 1 record
|
||||
count, err := client.db.Model(&model.Feed{}).Count()
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 1, count)
|
||||
|
||||
// Insert duplicated feed
|
||||
err = client.CreateFeed(feed)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Check no duplicates inserted
|
||||
count, err = client.db.Model(&model.Feed{}).Count()
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 1, count)
|
||||
}
|
||||
|
||||
func TestPgStorage_GetFeed(t *testing.T) {
|
||||
feed := &model.Feed{
|
||||
HashID: "xyz",
|
||||
UserID: "123",
|
||||
Provider: api.ProviderYoutube,
|
||||
LinkType: api.LinkTypeChannel,
|
||||
ItemID: "123",
|
||||
}
|
||||
|
||||
client := createClient(t)
|
||||
client.CreateFeed(feed)
|
||||
|
||||
out, err := client.GetFeed("xyz")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, feed.FeedID, out.FeedID)
|
||||
}
|
||||
|
||||
func TestPgStorage_UpdateLastAccess(t *testing.T) {
|
||||
feed := &model.Feed{
|
||||
HashID: "xyz",
|
||||
UserID: "123",
|
||||
Provider: api.ProviderYoutube,
|
||||
LinkType: api.LinkTypeChannel,
|
||||
ItemID: "123",
|
||||
}
|
||||
|
||||
client := createClient(t)
|
||||
err := client.CreateFeed(feed)
|
||||
require.NoError(t, err)
|
||||
|
||||
lastAccess := feed.LastAccess
|
||||
require.True(t, lastAccess.Unix() > 0)
|
||||
|
||||
last, err := client.GetFeed("xyz")
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NotEmpty(t, last.HashID)
|
||||
require.NotEmpty(t, last.UserID)
|
||||
require.NotEmpty(t, last.Provider)
|
||||
require.NotEmpty(t, last.LinkType)
|
||||
require.NotEmpty(t, last.ItemID)
|
||||
|
||||
require.True(t, last.LastAccess.UnixNano() > lastAccess.UnixNano())
|
||||
}
|
||||
|
||||
const TestDatabaseConnectionUrl = "postgres://postgres:@localhost/podsync?sslmode=disable"
|
||||
|
||||
func createClient(t *testing.T) *PgStorage {
|
||||
pg, err := NewPgStorage(&PgConfig{ConnectionUrl: TestDatabaseConnectionUrl})
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = pg.db.Model(&model.Feed{}).Where("1=1").Delete()
|
||||
require.NoError(t, err)
|
||||
|
||||
return pg
|
||||
}
|
||||
Reference in New Issue
Block a user