marco/yt-dlp-webui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

migrated to boltdb from sqlite + session files

Browse Source
This commit is contained in:
marcobaobao
2025-08-31 20:58:54 +02:00
parent 4c35b0b41f
commit 658d43f9ea
15 changed files with 448 additions and 400 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
server/internal/kv/store.go
View File

@@ -2,31 +2,58 @@ package kv
import (
"encoding/gob"
"encoding/json"
"errors"
"log/slog"
"os"
"path/filepath"
"runtime"
"sync"
"time"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/config"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/downloaders"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/queue"
bolt "go.etcd.io/bbolt"
)
var memDbEvents = make(chan downloaders.Downloader, runtime.NumCPU())
var (
bucket = []byte("downloads")
memDbEvents = make(chan downloaders.Downloader, runtime.NumCPU())
)
// In-Memory Thread-Safe Key-Value Storage with optional persistence
type Store struct {
db *bolt.DB
table map[string]downloaders.Downloader
mu sync.RWMutex
}
func NewStore() *Store {
return &Store{
func NewStore(db *bolt.DB, snaptshotInteval time.Duration) (*Store, error) {
// init bucket
err := db.Update(func(tx *bolt.Tx) error {
_, err := tx.CreateBucketIfNotExists(bucket)
return err
})
if err != nil {
return nil, err
}
s := &Store{
db: db,
table: make(map[string]downloaders.Downloader),
}
go func() {
ticker := time.NewTicker(snaptshotInteval)
for range ticker.C {
s.Snapshot()
}
}()
return s, err
}
// Get a process pointer given its id
@@ -108,25 +135,25 @@ func (m *Store) Persist() error {
// Restore a persisted state
func (m *Store) Restore(mq *queue.MessageQueue) {
sf := filepath.Join(config.Instance().SessionFilePath, "session.dat")
fd, err := os.Open(sf)
if err != nil {
return
}
var session Session
if err := gob.NewDecoder(fd).Decode(&session); err != nil {
return
}
m.mu.Lock()
defer m.mu.Unlock()
for _, snap := range session.Processes {
var restored downloaders.Downloader
var snapshot []internal.ProcessSnapshot
m.db.View(func(tx *bolt.Tx) error {
b := tx.Bucket(bucket)
return b.ForEach(func(k, v []byte) error {
var snap internal.ProcessSnapshot
if err := json.Unmarshal(v, &snap); err != nil {
return err
}
snapshot = append(snapshot, snap)
return nil
})
})
for _, snap := range snapshot {
var restored downloaders.Downloader
if snap.DownloaderName == "generic" {
d := downloaders.NewGenericDownload("", []string{})
err := d.RestoreFromSnapshot(&snap)
@@ -134,9 +161,7 @@ func (m *Store) Restore(mq *queue.MessageQueue) {
continue
}
restored = d
m.table[snap.Id] = restored
if !restored.(*downloaders.GenericDownloader).DownloaderBase.Completed {
mq.Publish(restored)
}
@@ -152,3 +177,23 @@ func (m *Store) EventListener() {
}
}
}
func (m *Store) Snapshot() error {
slog.Debug("snapshotting downloads state")
running := m.All()
return m.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket(bucket)
for _, v := range *running {
data, err := json.Marshal(v)
if err != nil {
return err
}
if err := b.Put([]byte(v.Id), data); err != nil {
return err
}
}
return nil
})
}

10
server/internal/livestream/livestream.go
View File

@@ -35,11 +35,11 @@ type LiveStream struct {
waitTime time.Duration
liveDate time.Time
mq *queue.MessageQueue
db *kv.Store
mq *queue.MessageQueue
store *kv.Store
}
func New(url string, done chan *LiveStream, mq *queue.MessageQueue, db *kv.Store) *LiveStream {
func New(url string, done chan *LiveStream, mq *queue.MessageQueue, store *kv.Store) *LiveStream {
return &LiveStream{
url: url,
done: done,
@@ -47,7 +47,7 @@ func New(url string, done chan *LiveStream, mq *queue.MessageQueue, db *kv.Store
waitTime: time.Second * 0,
waitTimeChan: make(chan time.Duration),
mq: mq,
db: db,
store: store,
}
}
@@ -94,7 +94,7 @@ func (l *LiveStream) Start() error {
//TODO: add pipes
d := downloaders.NewLiveStreamDownloader(l.url, []pipes.Pipe{})
l.db.Set(d)
l.store.Set(d)
l.mq.Publish(d)
return nil

80
server/internal/livestream/monitor.go
View File

@@ -1,28 +1,26 @@
package livestream
import (
"encoding/gob"
"log/slog"
"maps"
"os"
"path/filepath"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/config"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/kv"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/queue"
bolt "go.etcd.io/bbolt"
)
var bucket = []byte("livestreams")
type Monitor struct {
db *kv.Store // where the just started livestream will be published
db *bolt.DB
store *kv.Store // where the just started livestream will be published
mq *queue.MessageQueue // where the just started livestream will be published
streams map[string]*LiveStream // keeps track of the livestreams
done chan *LiveStream // to signal individual processes completition
}
func NewMonitor(mq *queue.MessageQueue, db *kv.Store) *Monitor {
func NewMonitor(mq *queue.MessageQueue, store *kv.Store, db *bolt.DB) *Monitor {
return &Monitor{
mq: mq,
db: db,
store: store,
streams: make(map[string]*LiveStream),
done: make(chan *LiveStream),
}
@@ -32,14 +30,24 @@ func NewMonitor(mq *queue.MessageQueue, db *kv.Store) *Monitor {
func (m *Monitor) Schedule() {
for l := range m.done {
delete(m.streams, l.url)
m.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket(bucket)
return b.Delete([]byte(l.url))
})
}
}
func (m *Monitor) Add(url string) {
ls := New(url, m.done, m.mq, m.db)
ls := New(url, m.done, m.mq, m.store)
go ls.Start()
m.streams[url] = ls
m.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket(bucket)
return b.Put([]byte(url), []byte{})
})
}
func (m *Monitor) Remove(url string) error {
@@ -59,11 +67,6 @@ func (m *Monitor) Status() LiveStreamStatus {
status := make(LiveStreamStatus)
for k, v := range m.streams {
// wt, ok := <-v.WaitTime()
// if !ok {
// continue
// }
status[k] = Status{
Status: v.status,
WaitTime: v.waitTime,
@@ -74,46 +77,13 @@ func (m *Monitor) Status() LiveStreamStatus {
return status
}
// Persist the monitor current state to a file.
// The file is located in the configured config directory
func (m *Monitor) Persist() error {
fd, err := os.Create(filepath.Join(config.Instance().SessionFilePath, "livestreams.dat"))
if err != nil {
return err
}
defer fd.Close()
slog.Debug("persisting livestream monitor state")
var toPersist []string
for url := range maps.Keys(m.streams) {
toPersist = append(toPersist, url)
}
return gob.NewEncoder(fd).Encode(toPersist)
}
// Restore a saved state and resume the monitored livestreams
func (m *Monitor) Restore() error {
fd, err := os.Open(filepath.Join(config.Instance().SessionFilePath, "livestreams.dat"))
if err != nil {
return err
}
defer fd.Close()
var toRestore []string
if err := gob.NewDecoder(fd).Decode(&toRestore); err != nil {
return err
}
for _, url := range toRestore {
m.Add(url)
}
slog.Debug("restored livestream monitor state")
return nil
return m.db.View(func(tx *bolt.Tx) error {
b := tx.Bucket(bucket)
return b.ForEach(func(k, v []byte) error {
m.Add(string(k))
return nil
})
})
}

95
server/internal/pipeline/store.go Normal file
View File

@@ -0,0 +1,95 @@
package pipeline
import (
"encoding/json"
"fmt"
bolt "go.etcd.io/bbolt"
)
var bucket = []byte("pipelines")
type Step struct {
Type string `json:"type"` // es. "transcoder", "filewriter"
FFmpegArgs []string `json:"ffmpeg_args,omitempty"` // args da passare a ffmpeg
Path string `json:"path,omitempty"` // solo per filewriter
}
type Pipeline struct {
ID string `json:"id"`
Name string `json:"name"`
Steps []Step `json:"steps"`
}
type Store struct {
db *bolt.DB
}
func NewStore(path string) (*Store, error) {
db, err := bolt.Open(path, 0600, nil)
if err != nil {
return nil, err
}
// init bucket
err = db.Update(func(tx *bolt.Tx) error {
_, err := tx.CreateBucketIfNotExists(bucket)
return err
})
if err != nil {
return nil, err
}
return &Store{db: db}, nil
}
func (s *Store) Save(p Pipeline) error {
data, err := json.Marshal(p)
if err != nil {
return err
}
return s.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket(bucket)
return b.Put([]byte(p.ID), data)
})
}
func (s *Store) Get(id string) (*Pipeline, error) {
var p Pipeline
err := s.db.View(func(tx *bolt.Tx) error {
b := tx.Bucket(bucket)
v := b.Get([]byte(id))
if v == nil {
return fmt.Errorf("pipeline %s not found", id)
}
return json.Unmarshal(v, &p)
})
if err != nil {
return nil, err
}
return &p, nil
}
func (s *Store) List() ([]Pipeline, error) {
var result []Pipeline
err := s.db.View(func(tx *bolt.Tx) error {
b := tx.Bucket(bucket)
return b.ForEach(func(k, v []byte) error {
var p Pipeline
if err := json.Unmarshal(v, &p); err != nil {
return err
}
result = append(result, p)
return nil
})
})
if err != nil {
return nil, err
}
return result, nil
}

2
server/middleware/utils.go
View File

@@ -11,9 +11,11 @@ func ApplyAuthenticationByConfig(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if config.Instance().RequireAuth {
Authenticated(next)
return
}
if config.Instance().UseOpenId {
openid.Middleware(next)
return
}
next.ServeHTTP(w, r)
})

8
server/rest/common.go
View File

@@ -1,14 +1,16 @@
package rest
import (
"database/sql"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/kv"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/livestream"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/queue"
bolt "go.etcd.io/bbolt"
)
type ContainerArgs struct {
DB *sql.DB
DB *bolt.DB
MDB *kv.Store
MQ *queue.MessageQueue
LM *livestream.Monitor
}

6
server/rest/provider.go
View File

@@ -14,11 +14,7 @@ var (
func ProvideService(args *ContainerArgs) *Service {
serviceOnce.Do(func() {
service = &Service{
mdb: args.MDB,
db: args.DB,
mq: args.MQ,
}
service = NewService(args.MDB, args.DB, args.MQ, args.LM)
})
return service
}

111
server/rest/service.go
View File

@@ -2,8 +2,9 @@ package rest
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"os/exec"
@@ -17,15 +18,35 @@ import (
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/livestream"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/queue"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/playlist"
bolt "go.etcd.io/bbolt"
)
type Service struct {
mdb *kv.Store
db *sql.DB
db *bolt.DB
mq *queue.MessageQueue
lm *livestream.Monitor
}
func NewService(
mdb *kv.Store,
db *bolt.DB,
mq *queue.MessageQueue,
lm *livestream.Monitor,
) *Service {
db.Update(func(tx *bolt.Tx) error {
_, err := tx.CreateBucketIfNotExists([]byte("templates"))
return err
})
return &Service{
mdb: mdb,
db: db,
mq: mq,
lm: lm,
}
}
func (s *Service) Exec(req internal.DownloadRequest) (string, error) {
d := downloaders.NewGenericDownload(req.URL, req.Params)
d.SetOutput(internal.DownloadOutput{
@@ -85,64 +106,56 @@ func (s *Service) SetCookies(ctx context.Context, cookies string) error {
}
func (s *Service) SaveTemplate(ctx context.Context, template *internal.CustomTemplate) error {
conn, err := s.db.Conn(ctx)
if err != nil {
return err
}
defer conn.Close()
_, err = conn.ExecContext(
ctx,
"INSERT INTO templates (id, name, content) VALUES (?, ?, ?)",
uuid.NewString(),
template.Name,
template.Content,
)
return err
return s.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte("templates"))
v, err := json.Marshal(template)
if err != nil {
return err
}
return b.Put([]byte(uuid.NewString()), v)
})
}
func (s *Service) GetTemplates(ctx context.Context) (*[]internal.CustomTemplate, error) {
conn, err := s.db.Conn(ctx)
if err != nil {
return nil, err
}
defer conn.Close()
rows, err := conn.QueryContext(ctx, "SELECT * FROM templates")
if err != nil {
return nil, err
}
defer rows.Close()
templates := make([]internal.CustomTemplate, 0)
for rows.Next() {
t := internal.CustomTemplate{}
err := rows.Scan(&t.Id, &t.Name, &t.Content)
if err != nil {
return nil, err
err := s.db.View(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte("templates"))
if b == nil {
return nil // bucket vuoto, restituisco lista vuota
}
templates = append(templates, t)
return b.ForEach(func(k, v []byte) error {
var t internal.CustomTemplate
if err := json.Unmarshal(v, &t); err != nil {
return err
}
templates = append(templates, t)
return nil
})
})
if err != nil {
return nil, err
}
return &templates, nil
}
func (s *Service) UpdateTemplate(ctx context.Context, t *internal.CustomTemplate) (*internal.CustomTemplate, error) {
conn, err := s.db.Conn(ctx)
data, err := json.Marshal(t)
if err != nil {
return nil, err
}
defer conn.Close()
err = s.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte("templates"))
if b == nil {
return fmt.Errorf("bucket templates not found")
}
return b.Put([]byte(t.Id), data)
})
_, err = conn.ExecContext(ctx, "UPDATE templates SET name = ?, content = ? WHERE id = ?", t.Name, t.Content, t.Id)
if err != nil {
return nil, err
}
@@ -151,16 +164,10 @@ func (s *Service) UpdateTemplate(ctx context.Context, t *internal.CustomTemplate
}
func (s *Service) DeleteTemplate(ctx context.Context, id string) error {
conn, err := s.db.Conn(ctx)
if err != nil {
return err
}
defer conn.Close()
_, err = conn.ExecContext(ctx, "DELETE FROM templates WHERE id = ?", id)
return err
return s.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte("templates"))
return b.Delete([]byte(id))
})
}
func (s *Service) GetVersion(ctx context.Context) (string, string, error) {

67
server/server.go
View File

@@ -3,7 +3,6 @@ package server
import (
"context"
"database/sql"
"fmt"
"io"
"io/fs"
@@ -13,16 +12,14 @@ import (
"net/rpc"
"os"
"os/signal"
"path/filepath"
"strings"
"syscall"
"time"
"github.com/go-chi/chi/v5"
"github.com/go-chi/cors"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/archive"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/archiver"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/config"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/dbutil"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/filebrowser"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/kv"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/livestream"
@@ -38,7 +35,7 @@ import (
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/twitch"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/user"
_ "modernc.org/sqlite"
bolt "go.etcd.io/bbolt"
)
type RunConfig struct {
@@ -50,7 +47,7 @@ type serverConfig struct {
frontend fs.FS
swagger fs.FS
mdb *kv.Store
db *sql.DB
db *bolt.DB
mq *queue.MessageQueue
lm *livestream.Monitor
tm *twitch.Monitor
@@ -60,7 +57,17 @@ type serverConfig struct {
var observableLogger = logging.NewObservableLogger()
func RunBlocking(rc *RunConfig) {
mdb := kv.NewStore()
dbPath := filepath.Join(config.Instance().SessionFilePath, "bolt.db")
boltdb, err := bolt.Open(dbPath, 0600, nil)
if err != nil {
panic(err)
}
mdb, err := kv.NewStore(boltdb, time.Second*15)
if err != nil {
panic(err)
}
// ---- LOGGING ---------------------------------------------------
logWriters := []io.Writer{
@@ -97,15 +104,6 @@ func RunBlocking(rc *RunConfig) {
slog.SetDefault(logger)
// ----------------------------------------------------------------
db, err := sql.Open("sqlite", conf.LocalDatabasePath)
if err != nil {
slog.Error("failed to open database", slog.String("err", err.Error()))
}
if err := dbutil.Migrate(context.Background(), db); err != nil {
slog.Error("failed to init database", slog.String("err", err.Error()))
}
mq, err := queue.NewMessageQueue()
if err != nil {
panic(err)
@@ -114,7 +112,7 @@ func RunBlocking(rc *RunConfig) {
go mdb.Restore(mq)
go mdb.EventListener()
lm := livestream.NewMonitor(mq, mdb)
lm := livestream.NewMonitor(mq, mdb, boltdb)
go lm.Schedule()
go lm.Restore()
@@ -123,6 +121,7 @@ func RunBlocking(rc *RunConfig) {
config.Instance().Twitch.ClientId,
config.Instance().Twitch.ClientSecret,
),
boltdb,
)
go tm.Monitor(
context.TODO(),
@@ -135,8 +134,8 @@ func RunBlocking(rc *RunConfig) {
frontend: rc.App,
swagger: rc.Swagger,
mdb: mdb,
db: boltdb,
mq: mq,
db: db,
lm: lm,
tm: tm,
}
@@ -144,7 +143,6 @@ func RunBlocking(rc *RunConfig) {
srv := newServer(scfg)
go gracefulShutdown(srv, &scfg)
go autoPersist(time.Minute*5, mdb, lm, tm)
var (
network = "tcp"
@@ -171,7 +169,7 @@ func RunBlocking(rc *RunConfig) {
}
func newServer(c serverConfig) *http.Server {
archiver.Register(c.db)
// archiver.Register(c.db)
cronTaskRunner := task.NewCronTaskRunner(c.mq, c.mdb)
go cronTaskRunner.Spawner(context.TODO())
@@ -216,7 +214,7 @@ func newServer(c serverConfig) *http.Server {
})
// Archive routes
r.Route("/archive", archive.ApplyRouter(c.db))
// r.Route("/archive", archive.ApplyRouter(c.db))
// Authentication routes
r.Route("/auth", func(r chi.Router) {
@@ -238,6 +236,7 @@ func newServer(c serverConfig) *http.Server {
DB: c.db,
MDB: c.mdb,
MQ: c.mq,
LM: c.lm,
}))
// Logging
@@ -273,34 +272,10 @@ func gracefulShutdown(srv *http.Server, cfg *serverConfig) {
defer func() {
cfg.mdb.Persist()
cfg.lm.Persist()
cfg.tm.Persist()
cfg.db.Close()
stop()
srv.Shutdown(context.Background())
}()
}()
}
func autoPersist(
d time.Duration,
db *kv.Store,
lm *livestream.Monitor,
tm *twitch.Monitor,
) {
for {
time.Sleep(d)
if err := db.Persist(); err != nil {
slog.Warn("failed to persisted session", slog.Any("err", err))
}
if err := lm.Persist(); err != nil {
slog.Warn(
"failed to persisted livestreams monitor session", slog.Any("err", err.Error()))
}
if err := tm.Persist(); err != nil {
slog.Warn(
"failed to persisted twitch monitor session", slog.Any("err", err.Error()))
}
slog.Debug("sucessfully persisted session")
}
}

6
server/subscription/container.go
View File

@@ -1,13 +1,13 @@
package subscription
import (
"database/sql"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/subscription/domain"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/subscription/task"
bolt "go.etcd.io/bbolt"
)
func Container(db *sql.DB, runner task.TaskRunner) domain.RestHandler {
func Container(db *bolt.DB, runner task.TaskRunner) domain.RestHandler {
var (
r = provideRepository(db)
s = provideService(r, runner)

5
server/subscription/provider.go
View File

@@ -1,7 +1,6 @@
package subscription
import (
"database/sql"
"sync"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/subscription/domain"
@@ -9,6 +8,8 @@ import (
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/subscription/rest"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/subscription/service"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/subscription/task"
bolt "go.etcd.io/bbolt"
)
var (
@@ -21,7 +22,7 @@ var (
handOnce sync.Once
)
func provideRepository(db *sql.DB) domain.Repository {
func provideRepository(db *bolt.DB) domain.Repository {
repoOnce.Do(func() {
repo = repository.New(db)
})

175
server/subscription/repository/repository.go
View File

@@ -2,131 +2,142 @@ package repository
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"github.com/google/uuid"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/subscription/data"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/subscription/domain"
bolt "go.etcd.io/bbolt"
)
var bucketName = []byte("subscriptions")
type Repository struct {
db *sql.DB
db *bolt.DB
}
// Delete implements domain.Repository.
func (r *Repository) Delete(ctx context.Context, id string) error {
conn, err := r.db.Conn(ctx)
if err != nil {
return err
}
defer conn.Close()
_, err = conn.ExecContext(ctx, "DELETE FROM subscriptions WHERE id = ?", id)
return err
return r.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket(bucketName)
return b.Delete([]byte(id))
})
}
// GetCursor implements domain.Repository.
func (r *Repository) GetCursor(ctx context.Context, id string) (int64, error) {
conn, err := r.db.Conn(ctx)
func (s *Repository) GetCursor(ctx context.Context, id string) (int64, error) {
var cursor int64
err := s.db.View(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte("subscriptions"))
v := b.Get([]byte(id))
if v == nil {
return fmt.Errorf("subscription %s not found", id)
}
var data struct {
Cursor int64 `json:"cursor"`
}
if err := json.Unmarshal(v, &data); err != nil {
return err
}
cursor = data.Cursor
return nil
})
if err != nil {
return -1, err
}
defer conn.Close()
row := conn.QueryRowContext(ctx, "SELECT rowid FROM subscriptions WHERE id = ?", id)
var rowId int64
if err := row.Scan(&rowId); err != nil {
return -1, err
}
return rowId, nil
return cursor, nil
}
// List implements domain.Repository.
func (r *Repository) List(ctx context.Context, start int64, limit int) (*[]data.Subscription, error) {
conn, err := r.db.Conn(ctx)
var subs []data.Subscription
err := r.db.View(func(tx *bolt.Tx) error {
b := tx.Bucket(bucketName)
return b.ForEach(func(k, v []byte) error {
var sub data.Subscription
if err := json.Unmarshal(v, &sub); err != nil {
return err
}
subs = append(subs, sub)
return nil
})
})
if err != nil {
return nil, err
}
defer conn.Close()
var elements []data.Subscription
rows, err := conn.QueryContext(ctx, "SELECT rowid, * FROM subscriptions WHERE rowid > ? LIMIT ?", start, limit)
if err != nil {
return nil, err
}
for rows.Next() {
var rowId int64
var element data.Subscription
if err := rows.Scan(
&rowId,
&element.Id,
&element.URL,
&element.Params,
&element.CronExpr,
); err != nil {
return &elements, err
}
elements = append(elements, element)
}
return &elements, nil
return &subs, nil
}
// Submit implements domain.Repository.
func (r *Repository) Submit(ctx context.Context, sub *data.Subscription) (*data.Subscription, error) {
conn, err := r.db.Conn(ctx)
func (s *Repository) Submit(ctx context.Context, sub *data.Subscription) (*data.Subscription, error) {
if sub.Id == "" {
sub.Id = uuid.NewString()
}
data, err := json.Marshal(sub)
if err != nil {
return nil, err
}
defer conn.Close()
err = s.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte("subscriptions"))
return b.Put([]byte(sub.Id), data)
})
_, err = conn.ExecContext(
ctx,
"INSERT INTO subscriptions (id, url, params, cron) VALUES (?, ?, ?, ?)",
uuid.NewString(),
sub.URL,
sub.Params,
sub.CronExpr,
)
if err != nil {
return nil, err
}
return sub, err
return sub, nil
}
// UpdateByExample implements domain.Repository.
func (r *Repository) UpdateByExample(ctx context.Context, example *data.Subscription) error {
conn, err := r.db.Conn(ctx)
if err != nil {
return err
}
func (s *Repository) UpdateByExample(ctx context.Context, example *data.Subscription) error {
return s.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte("subscriptions"))
defer conn.Close()
return b.ForEach(func(k, v []byte) error {
var sub data.Subscription
if err := json.Unmarshal(v, &sub); err != nil {
return err
}
_, err = conn.ExecContext(
ctx,
"UPDATE subscriptions SET url = ?, params = ?, cron = ? WHERE id = ? OR url = ?",
example.URL,
example.Params,
example.CronExpr,
example.Id,
example.URL,
)
if sub.Id == example.Id || sub.URL == example.URL {
// aggiorna i campi
sub.URL = example.URL
sub.Params = example.Params
sub.CronExpr = example.CronExpr
return err
data, err := json.Marshal(sub)
if err != nil {
return err
}
if err := b.Put(k, data); err != nil {
return err
}
}
return nil
})
})
}
func New(db *sql.DB) domain.Repository {
func New(db *bolt.DB) domain.Repository {
db.Update(func(tx *bolt.Tx) error {
_, err := tx.CreateBucketIfNotExists(bucketName)
return err
})
return &Repository{
db: db,
}

79
server/twitch/monitor.go
View File

@@ -2,12 +2,10 @@ package twitch
import (
"context"
"encoding/gob"
"fmt"
"iter"
"log/slog"
"maps"
"os"
"path/filepath"
"sync"
"time"
@@ -17,29 +15,48 @@ import (
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/kv"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/pipes"
"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal/queue"
bolt "go.etcd.io/bbolt"
)
var bucket = []byte("twitch-monitor")
type Monitor struct {
liveChannel chan *StreamInfo
monitored map[string]*Client
lastState map[string]bool
mu sync.RWMutex
db *bolt.DB
authenticationManager *AuthenticationManager
}
func NewMonitor(authenticationManager *AuthenticationManager) *Monitor {
func NewMonitor(authenticationManager *AuthenticationManager, db *bolt.DB) *Monitor {
db.Update(func(tx *bolt.Tx) error {
_, err := tx.CreateBucketIfNotExists(bucket)
return err
})
return &Monitor{
liveChannel: make(chan *StreamInfo, 16),
monitored: make(map[string]*Client),
lastState: make(map[string]bool),
authenticationManager: authenticationManager,
db: db,
}
}
func (m *Monitor) Add(user string) {
m.mu.Lock()
defer m.mu.Unlock()
m.monitored[user] = NewTwitchClient(m.authenticationManager)
m.mu.Unlock()
m.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket(bucket)
//TODO: the empty byte array will be replaced with configs per user
err := b.Put([]byte(user), []byte(""))
return err
})
slog.Info("added user to twitch monitor", slog.String("user", user))
}
@@ -88,9 +105,15 @@ func (m *Monitor) GetMonitoredUsers() iter.Seq[string] {
func (m *Monitor) DeleteUser(user string) {
m.mu.Lock()
defer m.mu.Unlock()
delete(m.monitored, user)
delete(m.lastState, user)
m.mu.Unlock()
m.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket(bucket)
err := b.Delete([]byte(user))
return err
})
}
func DEFAULT_DOWNLOAD_HANDLER(db *kv.Store, mq *queue.MessageQueue) func(user string) error {
@@ -106,10 +129,6 @@ func DEFAULT_DOWNLOAD_HANDLER(db *kv.Store, mq *queue.MessageQueue) func(user st
)
d := downloaders.NewLiveStreamDownloader(url, []pipes.Pipe{
// &pipes.FileWriter{
// Path: filename + ".mp4",
// IsFinal: false,
// },
&pipes.Transcoder{
Args: []string{
"-c:a", "libopus",
@@ -130,42 +149,16 @@ func DEFAULT_DOWNLOAD_HANDLER(db *kv.Store, mq *queue.MessageQueue) func(user st
}
}
func (m *Monitor) Persist() error {
filename := filepath.Join(config.Instance().SessionFilePath, "twitch-monitor.dat")
f, err := os.Create(filename)
if err != nil {
return err
}
defer f.Close()
enc := gob.NewEncoder(f)
users := make([]string, 0, len(m.monitored))
for user := range m.monitored {
users = append(users, user)
}
return enc.Encode(users)
}
func (m *Monitor) Restore() error {
filename := filepath.Join(config.Instance().SessionFilePath, "twitch-monitor.dat")
f, err := os.Open(filename)
if err != nil {
if os.IsNotExist(err) {
return nil
}
return err
}
defer f.Close()
dec := gob.NewDecoder(f)
var users []string
if err := dec.Decode(&users); err != nil {
return err
}
m.db.View(func(tx *bolt.Tx) error {
b := tx.Bucket(bucket)
return b.ForEach(func(k, v []byte) error {
users = append(users, string(k))
return nil
})
})
m.monitored = make(map[string]*Client)
for _, user := range users {