feat(history): persist topics and messages in SQLite
Store topics, messages, and small UI preferences in $XDG_DATA_HOME/lazyntfy/history.db. Each topic keeps its newest 1000 messages, expired rows are pruned on startup, and credentials are never written. Saves use a single transaction and prune only when a topic is over the cap.
This commit is contained in:
@@ -0,0 +1,262 @@
|
|||||||
|
// Package history stores topics and messages on disk so they survive restarts.
|
||||||
|
// Credentials are not written.
|
||||||
|
package history
|
||||||
|
|
||||||
|
import (
|
||||||
|
"database/sql"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
_ "modernc.org/sqlite"
|
||||||
|
|
||||||
|
"lazyntfy/internal/config"
|
||||||
|
"lazyntfy/internal/store"
|
||||||
|
)
|
||||||
|
|
||||||
|
const maxPerTopic = 1000
|
||||||
|
|
||||||
|
// DB is the on-disk history.
|
||||||
|
type DB struct {
|
||||||
|
sql *sql.DB
|
||||||
|
}
|
||||||
|
|
||||||
|
// DefaultPath is $XDG_DATA_HOME/lazyntfy/history.db, or ~/.local/share/lazyntfy/history.db.
|
||||||
|
func DefaultPath() string {
|
||||||
|
base := os.Getenv("XDG_DATA_HOME")
|
||||||
|
if base == "" {
|
||||||
|
home, err := os.UserHomeDir()
|
||||||
|
if err != nil {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
base = filepath.Join(home, ".local", "share")
|
||||||
|
}
|
||||||
|
return filepath.Join(base, "lazyntfy", "history.db")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Open creates the database and its tables.
|
||||||
|
func Open(path string) (*DB, error) {
|
||||||
|
if path == "" {
|
||||||
|
return nil, fmt.Errorf("empty history path")
|
||||||
|
}
|
||||||
|
if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
db, err := sql.Open("sqlite", path)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
db.SetMaxOpenConns(1)
|
||||||
|
if _, err := db.Exec(`
|
||||||
|
PRAGMA journal_mode = WAL;
|
||||||
|
PRAGMA busy_timeout = 5000;
|
||||||
|
CREATE TABLE IF NOT EXISTS topics (
|
||||||
|
origin TEXT NOT NULL,
|
||||||
|
name TEXT NOT NULL,
|
||||||
|
display TEXT NOT NULL DEFAULT '',
|
||||||
|
PRIMARY KEY (origin, name)
|
||||||
|
);
|
||||||
|
CREATE TABLE IF NOT EXISTS messages (
|
||||||
|
id TEXT PRIMARY KEY,
|
||||||
|
origin TEXT NOT NULL DEFAULT '',
|
||||||
|
topic TEXT NOT NULL,
|
||||||
|
time INTEGER NOT NULL,
|
||||||
|
expires INTEGER NOT NULL DEFAULT 0,
|
||||||
|
title TEXT NOT NULL DEFAULT '',
|
||||||
|
body TEXT NOT NULL DEFAULT '',
|
||||||
|
tags TEXT NOT NULL DEFAULT '[]',
|
||||||
|
priority INTEGER NOT NULL DEFAULT 0,
|
||||||
|
click TEXT NOT NULL DEFAULT '',
|
||||||
|
attachment TEXT NOT NULL DEFAULT '',
|
||||||
|
actions TEXT NOT NULL DEFAULT ''
|
||||||
|
);
|
||||||
|
CREATE INDEX IF NOT EXISTS messages_topic_time ON messages (topic, time DESC);
|
||||||
|
CREATE TABLE IF NOT EXISTS state (
|
||||||
|
key TEXT PRIMARY KEY,
|
||||||
|
value TEXT NOT NULL
|
||||||
|
);
|
||||||
|
`); err != nil {
|
||||||
|
_ = db.Close()
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
h := &DB{sql: db}
|
||||||
|
h.pruneExpired()
|
||||||
|
return h, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Close releases the file.
|
||||||
|
func (d *DB) Close() error {
|
||||||
|
if d == nil || d.sql == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return d.sql.Close()
|
||||||
|
}
|
||||||
|
|
||||||
|
// SaveTopic remembers a topic name. Auth is intentionally omitted.
|
||||||
|
func (d *DB) SaveTopic(sub config.Sub) error {
|
||||||
|
if d == nil || sub.Name == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
_, err := d.sql.Exec(`
|
||||||
|
INSERT INTO topics (origin, name, display) VALUES (?, ?, ?)
|
||||||
|
ON CONFLICT (origin, name) DO UPDATE SET display = excluded.display
|
||||||
|
`, sub.Origin, sub.Name, sub.Display)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Topics returns every remembered topic, newest names last as stored.
|
||||||
|
func (d *DB) Topics() ([]config.Sub, error) {
|
||||||
|
if d == nil {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
rows, err := d.sql.Query(`SELECT origin, name, display FROM topics ORDER BY name`)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var out []config.Sub
|
||||||
|
for rows.Next() {
|
||||||
|
var sub config.Sub
|
||||||
|
if err := rows.Scan(&sub.Origin, &sub.Name, &sub.Display); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out = append(out, sub)
|
||||||
|
}
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
// SaveMessage inserts or replaces one message and trims the topic to the cap.
|
||||||
|
func (d *DB) SaveMessage(origin string, m store.Message) error {
|
||||||
|
if d == nil || m.ID == "" || m.Topic == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
tags, err := json.Marshal(m.Tags)
|
||||||
|
if err != nil {
|
||||||
|
tags = []byte("[]")
|
||||||
|
}
|
||||||
|
attachment := ""
|
||||||
|
if m.Attachment != nil {
|
||||||
|
raw, err := json.Marshal(m.Attachment)
|
||||||
|
if err == nil {
|
||||||
|
attachment = string(raw)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
tx, err := d.sql.Begin()
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer func() { _ = tx.Rollback() }()
|
||||||
|
if _, err := tx.Exec(`
|
||||||
|
INSERT INTO messages (id, origin, topic, time, expires, title, body, tags, priority, click, attachment, actions)
|
||||||
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
ON CONFLICT (id) DO UPDATE SET
|
||||||
|
title = excluded.title,
|
||||||
|
body = excluded.body,
|
||||||
|
tags = excluded.tags,
|
||||||
|
priority = excluded.priority,
|
||||||
|
click = excluded.click,
|
||||||
|
attachment = excluded.attachment,
|
||||||
|
actions = excluded.actions
|
||||||
|
`, m.ID, origin, m.Topic, m.Time, m.Expires, m.Title, m.Body, string(tags), m.Priority, m.Click, attachment, m.Actions); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
var kept int
|
||||||
|
if err := tx.QueryRow(`SELECT COUNT(*) FROM messages WHERE topic = ?`, m.Topic).Scan(&kept); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if kept > maxPerTopic {
|
||||||
|
if _, err := tx.Exec(`
|
||||||
|
DELETE FROM messages WHERE topic = ? AND id NOT IN (
|
||||||
|
SELECT id FROM messages WHERE topic = ? ORDER BY time DESC LIMIT ?
|
||||||
|
)
|
||||||
|
`, m.Topic, m.Topic, maxPerTopic); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return tx.Commit()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Messages returns up to limit messages for a topic, newest first.
|
||||||
|
func (d *DB) Messages(topic string, limit int) ([]store.Message, error) {
|
||||||
|
if d == nil || topic == "" {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
if limit <= 0 {
|
||||||
|
limit = maxPerTopic
|
||||||
|
}
|
||||||
|
rows, err := d.sql.Query(`
|
||||||
|
SELECT id, topic, time, expires, title, body, tags, priority, click, attachment, actions
|
||||||
|
FROM messages WHERE topic = ? ORDER BY time DESC LIMIT ?
|
||||||
|
`, topic, limit)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var out []store.Message
|
||||||
|
for rows.Next() {
|
||||||
|
var m store.Message
|
||||||
|
var tags, attachment string
|
||||||
|
if err := rows.Scan(&m.ID, &m.Topic, &m.Time, &m.Expires, &m.Title, &m.Body, &tags, &m.Priority, &m.Click, &attachment, &m.Actions); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
_ = json.Unmarshal([]byte(tags), &m.Tags)
|
||||||
|
if attachment != "" {
|
||||||
|
var a store.Attachment
|
||||||
|
if err := json.Unmarshal([]byte(attachment), &a); err == nil && (a.Name != "" || a.URL != "") {
|
||||||
|
m.Attachment = &a
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out = append(out, m)
|
||||||
|
}
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeleteMessage removes one message.
|
||||||
|
func (d *DB) DeleteMessage(id string) error {
|
||||||
|
if d == nil || id == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
_, err := d.sql.Exec(`DELETE FROM messages WHERE id = ?`, id)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// ClearTopic removes every message for a topic.
|
||||||
|
func (d *DB) ClearTopic(topic string) error {
|
||||||
|
if d == nil || topic == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
_, err := d.sql.Exec(`DELETE FROM messages WHERE topic = ?`, topic)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetState stores a small UI preference, such as the last topic.
|
||||||
|
func (d *DB) SetState(key, value string) error {
|
||||||
|
if d == nil || key == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
_, err := d.sql.Exec(`
|
||||||
|
INSERT INTO state (key, value) VALUES (?, ?)
|
||||||
|
ON CONFLICT (key) DO UPDATE SET value = excluded.value
|
||||||
|
`, key, value)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// State reads a preference. ok is false when it was never set.
|
||||||
|
func (d *DB) State(key string) (string, bool) {
|
||||||
|
if d == nil || key == "" {
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
var value string
|
||||||
|
err := d.sql.QueryRow(`SELECT value FROM state WHERE key = ?`, key).Scan(&value)
|
||||||
|
if err != nil {
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
return value, true
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *DB) pruneExpired() {
|
||||||
|
now := time.Now().Unix()
|
||||||
|
_, _ = d.sql.Exec(`DELETE FROM messages WHERE expires > 0 AND expires < ?`, now)
|
||||||
|
}
|
||||||
@@ -0,0 +1,124 @@
|
|||||||
|
package history
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"lazyntfy/internal/config"
|
||||||
|
"lazyntfy/internal/store"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestRoundTrip(t *testing.T) {
|
||||||
|
path := filepath.Join(t.TempDir(), "history.db")
|
||||||
|
db, err := Open(path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := db.SaveTopic(config.Sub{Origin: "https://ntfy.example", Name: "alerts", Display: "Alerts"}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
msg := store.Message{
|
||||||
|
ID: "m1", Topic: "alerts", Time: 20, Title: "Disk", Body: "full",
|
||||||
|
Tags: []string{"ops"}, Priority: 5, Click: "https://example.com",
|
||||||
|
Attachment: &store.Attachment{Name: "a.png", URL: "https://example.com/a.png", Size: 4},
|
||||||
|
}
|
||||||
|
if err := db.SaveMessage("https://ntfy.example", msg); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := db.SaveMessage("https://ntfy.example", store.Message{ID: "m0", Topic: "alerts", Time: 10, Body: "older"}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := db.SetState("topic", "alerts"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := db.Close(); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
db, err = Open(path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
topics, err := db.Topics()
|
||||||
|
if err != nil || len(topics) != 1 || topics[0].Display != "Alerts" || topics[0].Auth.Token != "" {
|
||||||
|
t.Fatalf("%v %+v", err, topics)
|
||||||
|
}
|
||||||
|
got, err := db.Messages("alerts", 10)
|
||||||
|
if err != nil || len(got) != 2 || got[0].ID != "m1" || got[0].Tags[0] != "ops" || got[0].Attachment.Name != "a.png" {
|
||||||
|
t.Fatalf("%v %+v", err, got)
|
||||||
|
}
|
||||||
|
if name, ok := db.State("topic"); !ok || name != "alerts" {
|
||||||
|
t.Fatalf("state %q %v", name, ok)
|
||||||
|
}
|
||||||
|
if err := db.DeleteMessage("m0"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := db.ClearTopic("alerts"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
left, err := db.Messages("alerts", 10)
|
||||||
|
if err != nil || len(left) != 0 {
|
||||||
|
t.Fatalf("%v %+v", err, left)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPruneExpiredOnOpen(t *testing.T) {
|
||||||
|
path := filepath.Join(t.TempDir(), "history.db")
|
||||||
|
db, err := Open(path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
now := time.Now().Unix()
|
||||||
|
for _, m := range []store.Message{
|
||||||
|
{ID: "live", Topic: "t", Time: 1},
|
||||||
|
{ID: "gone", Topic: "t", Time: 2, Expires: now - 60},
|
||||||
|
{ID: "later", Topic: "t", Time: 3, Expires: now + 3600},
|
||||||
|
} {
|
||||||
|
if err := db.SaveMessage("o", m); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := db.Close(); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
db, err = Open(path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
got, err := db.Messages("t", 10)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(got) != 2 || got[0].ID != "later" || got[1].ID != "live" {
|
||||||
|
t.Fatalf("kept %+v", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTopicCapKeepsNewest(t *testing.T) {
|
||||||
|
db, err := Open(filepath.Join(t.TempDir(), "history.db"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
for i := 0; i < maxPerTopic+5; i++ {
|
||||||
|
msg := store.Message{ID: fmt.Sprintf("m%d", i), Topic: "t", Time: int64(i)}
|
||||||
|
if err := db.SaveMessage("o", msg); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
got, err := db.Messages("t", maxPerTopic+100)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(got) != maxPerTopic {
|
||||||
|
t.Fatalf("kept %d, want %d", len(got), maxPerTopic)
|
||||||
|
}
|
||||||
|
if got[0].ID != fmt.Sprintf("m%d", maxPerTopic+4) {
|
||||||
|
t.Fatalf("newest is %s", got[0].ID)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user