Files
rsspipe/internal/db/db.go
T

698 lines
18 KiB
Go

package db
import (
"database/sql"
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
"time"
"git.pengzhan.dev/haopengzhan/rsspipe/internal/models"
"github.com/google/uuid"
_ "modernc.org/sqlite"
)
type DB struct {
*sql.DB
}
func InitDB(dbPath string) (*DB, error) {
dir := filepath.Dir(dbPath)
if err := os.MkdirAll(dir, 0755); err != nil {
return nil, fmt.Errorf("failed to create db directory: %w", err)
}
db, err := sql.Open("sqlite", dbPath)
if err != nil {
return nil, fmt.Errorf("failed to open sqlite db: %w", err)
}
db.SetMaxOpenConns(10)
db.SetMaxIdleConns(5)
db.SetConnMaxLifetime(time.Hour)
pragmas := []string{
"PRAGMA journal_mode = WAL;",
"PRAGMA busy_timeout = 5000;",
"PRAGMA synchronous = NORMAL;",
"PRAGMA foreign_keys = ON;",
}
for _, p := range pragmas {
if _, err := db.Exec(p); err != nil {
return nil, fmt.Errorf("failed to set pragma %s: %w", p, err)
}
}
database := &DB{DB: db}
if err := database.migrate(); err != nil {
return nil, fmt.Errorf("failed to run migrations: %w", err)
}
return database, nil
}
func (d *DB) migrate() error {
schema := `
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY,
username TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
created_at DATETIME NOT NULL,
updated_at DATETIME NOT NULL
);
CREATE TABLE IF NOT EXISTS categories (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
parent_id TEXT,
sort_order INTEGER DEFAULT 0,
created_at DATETIME NOT NULL,
FOREIGN KEY(parent_id) REFERENCES categories(id) ON DELETE SET NULL
);
CREATE TABLE IF NOT EXISTS feeds (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
feed_url TEXT UNIQUE NOT NULL,
site_url TEXT,
category_id TEXT,
is_virtual_digest BOOLEAN DEFAULT 0,
refresh_interval_minutes INTEGER DEFAULT 30,
timeout_seconds INTEGER DEFAULT 10,
etag TEXT DEFAULT '',
last_modified TEXT DEFAULT '',
last_fetched_at DATETIME,
fetch_error TEXT DEFAULT '',
created_at DATETIME NOT NULL,
FOREIGN KEY(category_id) REFERENCES categories(id) ON DELETE SET NULL
);
CREATE TABLE IF NOT EXISTS feed_entries (
id TEXT PRIMARY KEY,
feed_id TEXT NOT NULL,
guid TEXT NOT NULL,
url TEXT NOT NULL,
title TEXT NOT NULL,
author TEXT DEFAULT '',
content TEXT NOT NULL,
raw_content TEXT DEFAULT '',
published_at DATETIME NOT NULL,
is_read BOOLEAN DEFAULT 0,
is_starred BOOLEAN DEFAULT 0,
share_uuid TEXT UNIQUE,
metadata TEXT DEFAULT '{}',
created_at DATETIME NOT NULL,
FOREIGN KEY(feed_id) REFERENCES feeds(id) ON DELETE CASCADE,
UNIQUE(feed_id, guid)
);
CREATE INDEX IF NOT EXISTS idx_entries_feed_read ON feed_entries(feed_id, is_read);
CREATE INDEX IF NOT EXISTS idx_entries_published ON feed_entries(published_at DESC);
CREATE INDEX IF NOT EXISTS idx_entries_is_read ON feed_entries(is_read);
CREATE INDEX IF NOT EXISTS idx_entries_is_starred ON feed_entries(is_starred);
CREATE INDEX IF NOT EXISTS idx_entries_share_uuid ON feed_entries(share_uuid);
CREATE TABLE IF NOT EXISTS rules (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
rule_type TEXT NOT NULL,
is_active BOOLEAN DEFAULT 1,
trigger_type TEXT NOT NULL,
trigger_config TEXT DEFAULT '{}',
conditions TEXT DEFAULT '[]',
actions TEXT DEFAULT '[]',
created_at DATETIME NOT NULL,
updated_at DATETIME NOT NULL
);
`
if _, err := d.Exec(schema); err != nil {
return err
}
// Migration for timeout_seconds column if older DB exists
_, _ = d.Exec("ALTER TABLE feeds ADD COLUMN timeout_seconds INTEGER DEFAULT 10")
return nil
}
// User operations
func (d *DB) CountUsers() (int, error) {
var count int
err := d.QueryRow("SELECT COUNT(*) FROM users").Scan(&count)
return count, err
}
func (d *DB) GetUserByUsername(username string) (*models.User, error) {
var u models.User
err := d.QueryRow(
"SELECT id, username, password_hash, created_at, updated_at FROM users WHERE username = ?",
username,
).Scan(&u.ID, &u.Username, &u.PasswordHash, &u.CreatedAt, &u.UpdatedAt)
if err != nil {
return nil, err
}
return &u, nil
}
func (d *DB) CreateUser(username, passwordHash string) (*models.User, error) {
now := time.Now().UTC()
u := &models.User{
ID: uuid.New().String(),
Username: username,
PasswordHash: passwordHash,
CreatedAt: now,
UpdatedAt: now,
}
_, err := d.Exec(
"INSERT INTO users (id, username, password_hash, created_at, updated_at) VALUES (?, ?, ?, ?, ?)",
u.ID, u.Username, u.PasswordHash, u.CreatedAt, u.UpdatedAt,
)
if err != nil {
return nil, err
}
return u, nil
}
func (d *DB) UpdateUserPassword(userID, passwordHash string) error {
now := time.Now().UTC()
_, err := d.Exec(
"UPDATE users SET password_hash = ?, updated_at = ? WHERE id = ?",
passwordHash, now, userID,
)
return err
}
// Category operations
func (d *DB) ListCategories() ([]models.Category, error) {
query := `
SELECT c.id, c.title, c.parent_id, c.sort_order, c.created_at, COUNT(f.id) as feed_count
FROM categories c
LEFT JOIN feeds f ON f.category_id = c.id
GROUP BY c.id
ORDER BY c.sort_order ASC, c.title ASC
`
rows, err := d.Query(query)
if err != nil {
return nil, err
}
defer func() {
_ = rows.Close()
}()
var categories []models.Category
for rows.Next() {
var c models.Category
var parentID sql.NullString
if err := rows.Scan(&c.ID, &c.Title, &parentID, &c.SortOrder, &c.CreatedAt, &c.FeedCount); err != nil {
return nil, err
}
if parentID.Valid {
p := parentID.String
c.ParentID = &p
}
categories = append(categories, c)
}
return categories, nil
}
func (d *DB) GetCategoryByTitle(title string) (*models.Category, error) {
var c models.Category
var parentID sql.NullString
err := d.QueryRow("SELECT id, title, parent_id, sort_order, created_at FROM categories WHERE title = ?", title).
Scan(&c.ID, &c.Title, &parentID, &c.SortOrder, &c.CreatedAt)
if err != nil {
return nil, err
}
if parentID.Valid {
p := parentID.String
c.ParentID = &p
}
return &c, nil
}
func (d *DB) CreateCategory(title string, parentID *string) (*models.Category, error) {
c := &models.Category{
ID: uuid.New().String(),
Title: title,
ParentID: parentID,
SortOrder: 0,
CreatedAt: time.Now().UTC(),
}
var pID interface{}
if parentID != nil {
pID = *parentID
}
_, err := d.Exec("INSERT INTO categories (id, title, parent_id, sort_order, created_at) VALUES (?, ?, ?, ?, ?)",
c.ID, c.Title, pID, c.SortOrder, c.CreatedAt)
if err != nil {
return nil, err
}
return c, nil
}
func (d *DB) DeleteCategory(id string) error {
_, err := d.Exec("DELETE FROM categories WHERE id = ?", id)
return err
}
// Feed operations
func (d *DB) ListFeeds() ([]models.Feed, error) {
query := `
SELECT
f.id, f.title, f.feed_url, f.site_url, f.category_id,
COALESCE(c.title, '') as category_title,
f.is_virtual_digest, f.refresh_interval_minutes, COALESCE(f.timeout_seconds, 10),
f.etag, f.last_modified, f.last_fetched_at, f.fetch_error, f.created_at,
COALESCE(SUM(CASE WHEN e.is_read = 0 THEN 1 ELSE 0 END), 0) as unread_count,
COUNT(e.id) as total_count
FROM feeds f
LEFT JOIN categories c ON c.id = f.category_id
LEFT JOIN feed_entries e ON e.feed_id = f.id
GROUP BY f.id
ORDER BY f.title ASC
`
rows, err := d.Query(query)
if err != nil {
return nil, err
}
defer func() {
_ = rows.Close()
}()
var feeds []models.Feed
for rows.Next() {
var f models.Feed
var catID, siteURL, etag, lastMod, fetchErr sql.NullString
var lastFetched sql.NullTime
if err := rows.Scan(
&f.ID, &f.Title, &f.FeedURL, &siteURL, &catID, &f.CategoryTitle,
&f.IsVirtualDigest, &f.RefreshIntervalMinutes, &f.TimeoutSeconds,
&etag, &lastMod, &lastFetched, &fetchErr, &f.CreatedAt, &f.UnreadCount, &f.TotalCount,
); err != nil {
return nil, err
}
if catID.Valid {
cid := catID.String
f.CategoryID = &cid
}
if siteURL.Valid {
f.SiteURL = siteURL.String
}
if etag.Valid {
f.ETag = etag.String
}
if lastMod.Valid {
f.LastModified = lastMod.String
}
if lastFetched.Valid {
t := lastFetched.Time
f.LastFetchedAt = &t
}
if fetchErr.Valid {
f.FetchError = fetchErr.String
}
feeds = append(feeds, f)
}
return feeds, nil
}
func (d *DB) GetFeed(id string) (*models.Feed, error) {
query := `
SELECT
f.id, f.title, f.feed_url, f.site_url, f.category_id,
COALESCE(c.title, '') as category_title,
f.is_virtual_digest, f.refresh_interval_minutes, COALESCE(f.timeout_seconds, 10),
f.etag, f.last_modified, f.last_fetched_at, f.fetch_error, f.created_at,
COALESCE(SUM(CASE WHEN e.is_read = 0 THEN 1 ELSE 0 END), 0) as unread_count,
COUNT(e.id) as total_count
FROM feeds f
LEFT JOIN categories c ON c.id = f.category_id
LEFT JOIN feed_entries e ON e.feed_id = f.id
WHERE f.id = ?
GROUP BY f.id
`
var f models.Feed
var catID, siteURL, etag, lastMod, fetchErr sql.NullString
var lastFetched sql.NullTime
err := d.QueryRow(query, id).Scan(
&f.ID, &f.Title, &f.FeedURL, &siteURL, &catID, &f.CategoryTitle,
&f.IsVirtualDigest, &f.RefreshIntervalMinutes, &f.TimeoutSeconds,
&etag, &lastMod, &lastFetched, &fetchErr, &f.CreatedAt, &f.UnreadCount, &f.TotalCount,
)
if err != nil {
return nil, err
}
if catID.Valid {
cid := catID.String
f.CategoryID = &cid
}
if siteURL.Valid {
f.SiteURL = siteURL.String
}
if etag.Valid {
f.ETag = etag.String
}
if lastMod.Valid {
f.LastModified = lastMod.String
}
if lastFetched.Valid {
t := lastFetched.Time
f.LastFetchedAt = &t
}
if fetchErr.Valid {
f.FetchError = fetchErr.String
}
return &f, nil
}
func (d *DB) GetFeedByURL(feedURL string) (*models.Feed, error) {
var f models.Feed
var catID, siteURL, etag, lastMod, fetchErr sql.NullString
var lastFetched sql.NullTime
err := d.QueryRow(
"SELECT id, title, feed_url, site_url, category_id, is_virtual_digest, refresh_interval_minutes, COALESCE(timeout_seconds, 10), etag, last_modified, last_fetched_at, fetch_error, created_at FROM feeds WHERE feed_url = ?",
feedURL,
).Scan(
&f.ID, &f.Title, &f.FeedURL, &siteURL, &catID,
&f.IsVirtualDigest, &f.RefreshIntervalMinutes, &f.TimeoutSeconds, &etag, &lastMod,
&lastFetched, &fetchErr, &f.CreatedAt,
)
if err != nil {
return nil, err
}
if catID.Valid {
cid := catID.String
f.CategoryID = &cid
}
if siteURL.Valid {
f.SiteURL = siteURL.String
}
return &f, nil
}
func (d *DB) CreateFeed(title, feedURL, siteURL string, categoryID *string) (*models.Feed, error) {
now := time.Now().UTC()
f := &models.Feed{
ID: uuid.New().String(),
Title: title,
FeedURL: feedURL,
SiteURL: siteURL,
CategoryID: categoryID,
RefreshIntervalMinutes: 30,
TimeoutSeconds: 10,
CreatedAt: now,
}
var catID interface{}
if categoryID != nil {
catID = *categoryID
}
_, err := d.Exec(
"INSERT INTO feeds (id, title, feed_url, site_url, category_id, refresh_interval_minutes, timeout_seconds, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
f.ID, f.Title, f.FeedURL, f.SiteURL, catID, f.RefreshIntervalMinutes, f.TimeoutSeconds, f.CreatedAt,
)
if err != nil {
return nil, err
}
return f, nil
}
func (d *DB) UpdateFeed(id string, req models.UpdateFeedRequest) error {
feed, err := d.GetFeed(id)
if err != nil {
return err
}
title := feed.Title
if req.Title != nil && *req.Title != "" {
title = *req.Title
}
catID := feed.CategoryID
if req.CategoryID != nil {
if *req.CategoryID == "" {
catID = nil
} else {
catID = req.CategoryID
}
}
interval := feed.RefreshIntervalMinutes
if req.RefreshIntervalMinutes != nil && *req.RefreshIntervalMinutes > 0 {
interval = *req.RefreshIntervalMinutes
}
timeout := feed.TimeoutSeconds
if req.TimeoutSeconds != nil && *req.TimeoutSeconds > 0 {
timeout = *req.TimeoutSeconds
}
var cID interface{}
if catID != nil {
cID = *catID
}
_, err = d.Exec(`
UPDATE feeds
SET title = ?, category_id = ?, refresh_interval_minutes = ?, timeout_seconds = ?
WHERE id = ?
`, title, cID, interval, timeout, id)
return err
}
func (d *DB) UpdateFeedFetchStatus(id string, etag, lastModified, fetchErr string) error {
now := time.Now().UTC()
_, err := d.Exec(
"UPDATE feeds SET etag = ?, last_modified = ?, fetch_error = ?, last_fetched_at = ? WHERE id = ?",
etag, lastModified, fetchErr, now, id,
)
return err
}
func (d *DB) DeleteFeed(id string) error {
_, err := d.Exec("DELETE FROM feeds WHERE id = ?", id)
return err
}
// Feed Entries operations
func (d *DB) UpsertFeedEntries(feedID string, entries []models.FeedEntry) (int, error) {
if len(entries) == 0 {
return 0, nil
}
stmt, err := d.Prepare(`
INSERT INTO feed_entries (
id, feed_id, guid, url, title, author, content, raw_content, published_at, is_read, is_starred, metadata, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, ?, ?)
ON CONFLICT(feed_id, guid) DO UPDATE SET
title = excluded.title,
content = CASE WHEN feed_entries.content = '' THEN excluded.content ELSE feed_entries.content END,
url = excluded.url,
author = excluded.author
`)
if err != nil {
return 0, err
}
defer func() {
_ = stmt.Close()
}()
insertedCount := 0
for _, e := range entries {
id := uuid.New().String()
now := time.Now().UTC()
pubAt := e.PublishedAt
if pubAt.IsZero() {
pubAt = now
}
metaBytes, _ := json.Marshal(e.Metadata)
if len(metaBytes) == 0 {
metaBytes = []byte("{}")
}
res, err := stmt.Exec(
id, feedID, e.GUID, e.URL, e.Title, e.Author, e.Content, e.RawContent, pubAt, string(metaBytes), now,
)
if err == nil {
if n, _ := res.RowsAffected(); n > 0 {
insertedCount++
}
}
}
return insertedCount, nil
}
type EntryFilter struct {
View string // "inbox" (unread), "starred", "all"
FeedID string
CategoryID string
Search string
Limit int
Offset int
}
func (d *DB) ListEntries(f EntryFilter) ([]models.FeedEntry, int, error) {
where := []string{"1=1"}
args := []interface{}{}
switch f.View {
case "inbox":
where = append(where, "e.is_read = 0")
case "starred":
where = append(where, "e.is_starred = 1")
}
if f.FeedID != "" {
where = append(where, "e.feed_id = ?")
args = append(args, f.FeedID)
}
if f.CategoryID != "" {
where = append(where, "f.category_id = ?")
args = append(args, f.CategoryID)
}
if f.Search != "" {
searchTerms := strings.Fields(f.Search)
for _, term := range searchTerms {
lower := strings.ToLower(term)
if strings.HasPrefix(lower, "is:unread") {
where = append(where, "e.is_read = 0")
} else if strings.HasPrefix(lower, "is:read") {
where = append(where, "e.is_read = 1")
} else if strings.HasPrefix(lower, "is:starred") {
where = append(where, "e.is_starred = 1")
} else if strings.HasPrefix(lower, "feed:") {
feedVal := strings.TrimPrefix(term, "feed:")
where = append(where, "(f.title LIKE ? OR f.feed_url LIKE ?)")
args = append(args, "%"+feedVal+"%", "%"+feedVal+"%")
} else {
where = append(where, "(e.title LIKE ? OR e.content LIKE ? OR e.author LIKE ?)")
args = append(args, "%"+term+"%", "%"+term+"%", "%"+term+"%")
}
}
}
whereClause := strings.Join(where, " AND ")
// Count total
countQuery := fmt.Sprintf(`
SELECT COUNT(*)
FROM feed_entries e
JOIN feeds f ON f.id = e.feed_id
WHERE %s
`, whereClause)
var total int
if err := d.QueryRow(countQuery, args...).Scan(&total); err != nil {
return nil, 0, err
}
limit := f.Limit
if limit <= 0 || limit > 100 {
limit = 50
}
offset := f.Offset
if offset < 0 {
offset = 0
}
query := fmt.Sprintf(`
SELECT
e.id, e.feed_id, f.title as feed_title, COALESCE(f.site_url, '') as feed_site_url,
e.guid, e.url, e.title, e.author, e.content, e.raw_content,
e.published_at, e.is_read, e.is_starred, e.share_uuid, e.metadata, e.created_at
FROM feed_entries e
JOIN feeds f ON f.id = e.feed_id
WHERE %s
ORDER BY e.published_at DESC
LIMIT ? OFFSET ?
`, whereClause)
queryArgs := append(args, limit, offset)
rows, err := d.Query(query, queryArgs...)
if err != nil {
return nil, 0, err
}
defer func() {
_ = rows.Close()
}()
var entries []models.FeedEntry
for rows.Next() {
var e models.FeedEntry
var author, rawContent, metaStr sql.NullString
var shareUUID sql.NullString
if err := rows.Scan(
&e.ID, &e.FeedID, &e.FeedTitle, &e.FeedSiteURL,
&e.GUID, &e.URL, &e.Title, &author, &e.Content, &rawContent,
&e.PublishedAt, &e.IsRead, &e.IsStarred, &shareUUID, &metaStr, &e.CreatedAt,
); err != nil {
return nil, 0, err
}
if author.Valid {
e.Author = author.String
}
if rawContent.Valid {
e.RawContent = rawContent.String
}
if shareUUID.Valid {
s := shareUUID.String
e.ShareUUID = &s
}
e.Metadata = make(map[string]interface{})
if metaStr.Valid && metaStr.String != "" {
_ = json.Unmarshal([]byte(metaStr.String), &e.Metadata)
}
entries = append(entries, e)
}
return entries, total, nil
}
func (d *DB) UpdateEntryRead(id string, isRead bool) error {
_, err := d.Exec("UPDATE feed_entries SET is_read = ? WHERE id = ?", isRead, id)
return err
}
func (d *DB) UpdateEntryStarred(id string, isStarred bool) error {
_, err := d.Exec("UPDATE feed_entries SET is_starred = ? WHERE id = ?", isStarred, id)
return err
}
func (d *DB) MarkAllRead(feedID, categoryID *string) error {
if feedID != nil && *feedID != "" {
_, err := d.Exec("UPDATE feed_entries SET is_read = 1 WHERE feed_id = ?", *feedID)
return err
}
if categoryID != nil && *categoryID != "" {
_, err := d.Exec(`
UPDATE feed_entries
SET is_read = 1
WHERE feed_id IN (SELECT id FROM feeds WHERE category_id = ?)
`, *categoryID)
return err
}
_, err := d.Exec("UPDATE feed_entries SET is_read = 1")
return err
}
func (d *DB) GetStats() (*models.Stats, error) {
var s models.Stats
err := d.QueryRow("SELECT COUNT(*) FROM feeds").Scan(&s.TotalFeeds)
if err != nil {
return nil, err
}
err = d.QueryRow("SELECT COUNT(*) FROM feed_entries WHERE is_read = 0").Scan(&s.TotalUnread)
if err != nil {
return nil, err
}
err = d.QueryRow("SELECT COUNT(*) FROM feed_entries WHERE is_starred = 1").Scan(&s.TotalStarred)
if err != nil {
return nil, err
}
err = d.QueryRow("SELECT COUNT(*) FROM categories").Scan(&s.TotalCategories)
if err != nil {
return nil, err
}
return &s, nil
}