Store events and config in SQLite
Replace the JSON-file event and config stores with a SQLite database (modernc.org/sqlite, pure-Go so the static build keeps CGO_ENABLED=0). Last-write-wins now rides on the upsert's WHERE clause rather than a Go-side map compare; the sync protocol and HTTP handlers are unchanged. On first start the server auto-imports any legacy events.json/config.json sitting in the data dir, renaming them to *.imported. The -data flag now points at puppy.db; photos still live on the filesystem alongside it.
This commit is contained in:
Alexander Heldt
+202
-121
@@ -1,6 +1,7 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"flag"
|
||||
@@ -12,8 +13,9 @@ import (
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
type Event struct {
|
||||
@@ -45,155 +47,230 @@ type Config struct {
|
||||
}
|
||||
|
||||
type ConfigStore struct {
|
||||
path string
|
||||
mu sync.Mutex
|
||||
cfg Config
|
||||
db *sql.DB
|
||||
}
|
||||
|
||||
func newConfigStore(path string) (*ConfigStore, error) {
|
||||
cs := &ConfigStore{path: path}
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return cs, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
defer f.Close()
|
||||
if err := json.NewDecoder(f).Decode(&cs.cfg); err != nil && !errors.Is(err, io.EOF) {
|
||||
return nil, err
|
||||
}
|
||||
return cs, nil
|
||||
func newConfigStore(db *sql.DB) *ConfigStore {
|
||||
return &ConfigStore{db: db}
|
||||
}
|
||||
|
||||
func (cs *ConfigStore) get() Config {
|
||||
cs.mu.Lock()
|
||||
defer cs.mu.Unlock()
|
||||
return cs.cfg
|
||||
var c Config
|
||||
// The profile lives in a single row (id = 1). A missing row is the
|
||||
// pre-configuration state, so a zero-value Config is the right answer.
|
||||
err := cs.db.QueryRow(
|
||||
`SELECT name, birthday, updated FROM config WHERE id = 1`,
|
||||
).Scan(&c.Name, &c.Birthday, &c.UpdatedAt)
|
||||
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
||||
log.Printf("config get: %v", err)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
// merge applies an incoming config with last-write-wins by UpdatedAt and
|
||||
// returns the resulting stored config (which the caller sends back).
|
||||
func (cs *ConfigStore) merge(in Config) (Config, error) {
|
||||
cs.mu.Lock()
|
||||
defer cs.mu.Unlock()
|
||||
if in.UpdatedAt > cs.cfg.UpdatedAt {
|
||||
cs.cfg = in
|
||||
if err := cs.saveLocked(); err != nil {
|
||||
return cs.cfg, err
|
||||
}
|
||||
}
|
||||
return cs.cfg, nil
|
||||
}
|
||||
|
||||
// Caller must hold cs.mu.
|
||||
func (cs *ConfigStore) saveLocked() error {
|
||||
if err := os.MkdirAll(filepath.Dir(cs.path), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
tmp := cs.path + ".tmp"
|
||||
f, err := os.Create(tmp)
|
||||
// The upsert's WHERE clause enforces last-write-wins: the incoming row only
|
||||
// replaces the stored one when it is strictly newer.
|
||||
_, err := cs.db.Exec(`
|
||||
INSERT INTO config (id, name, birthday, updated)
|
||||
VALUES (1, ?, ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
name = excluded.name, birthday = excluded.birthday, updated = excluded.updated
|
||||
WHERE excluded.updated > config.updated`,
|
||||
in.Name, in.Birthday, in.UpdatedAt)
|
||||
if err != nil {
|
||||
return err
|
||||
return Config{}, err
|
||||
}
|
||||
enc := json.NewEncoder(f)
|
||||
enc.SetIndent("", " ")
|
||||
if err := enc.Encode(cs.cfg); err != nil {
|
||||
f.Close()
|
||||
os.Remove(tmp)
|
||||
return err
|
||||
}
|
||||
if err := f.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp, cs.path)
|
||||
return cs.get(), nil
|
||||
}
|
||||
|
||||
type Store struct {
|
||||
path string
|
||||
mu sync.Mutex
|
||||
data map[string]Event
|
||||
db *sql.DB
|
||||
}
|
||||
|
||||
func newStore(path string) (*Store, error) {
|
||||
s := &Store{path: path, data: map[string]Event{}}
|
||||
if err := s.load(); err != nil {
|
||||
func newStore(db *sql.DB) *Store {
|
||||
return &Store{db: db}
|
||||
}
|
||||
|
||||
// sync merges client events into the store using last-write-wins by UpdatedAt,
|
||||
// then returns the full merged set (including tombstones, which must propagate).
|
||||
func (s *Store) sync(client []Event) ([]Event, error) {
|
||||
tx, err := s.db.Begin()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s, nil
|
||||
defer tx.Rollback()
|
||||
|
||||
// The WHERE clause on the upsert is the last-write-wins rule: an incoming
|
||||
// event only overwrites the stored one when its updatedAt is strictly newer.
|
||||
stmt, err := tx.Prepare(`
|
||||
INSERT INTO events (id, type, at, note, photo_id, weight, updated, deleted)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
type = excluded.type, at = excluded.at, note = excluded.note,
|
||||
photo_id = excluded.photo_id, weight = excluded.weight,
|
||||
updated = excluded.updated, deleted = excluded.deleted
|
||||
WHERE excluded.updated > events.updated`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer stmt.Close()
|
||||
|
||||
for _, ce := range client {
|
||||
if ce.ID == "" {
|
||||
continue
|
||||
}
|
||||
if _, err := stmt.Exec(
|
||||
ce.ID, ce.Type, ce.At, ce.Note, ce.PhotoID, ce.Weight, ce.UpdatedAt, ce.Deleted,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.all()
|
||||
}
|
||||
|
||||
func (s *Store) load() error {
|
||||
f, err := os.Open(s.path)
|
||||
// all returns every stored event, tombstones included.
|
||||
func (s *Store) all() ([]Event, error) {
|
||||
rows, err := s.db.Query(
|
||||
`SELECT id, type, at, note, photo_id, weight, updated, deleted FROM events`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := make([]Event, 0)
|
||||
for rows.Next() {
|
||||
var e Event
|
||||
if err := rows.Scan(
|
||||
&e.ID, &e.Type, &e.At, &e.Note, &e.PhotoID, &e.Weight, &e.UpdatedAt, &e.Deleted,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, e)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// openDB opens (creating if needed) the SQLite database and ensures the schema
|
||||
// exists. WAL mode plays nicely with concurrent readers during a sync write;
|
||||
// busy_timeout avoids spurious "database is locked" errors under contention.
|
||||
func openDB(path string) (*sql.DB, error) {
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
db, err := sql.Open("sqlite", path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, pragma := range []string{
|
||||
`PRAGMA journal_mode = WAL`,
|
||||
`PRAGMA busy_timeout = 5000`,
|
||||
`PRAGMA synchronous = NORMAL`,
|
||||
} {
|
||||
if _, err := db.Exec(pragma); err != nil {
|
||||
db.Close()
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
schema := `
|
||||
CREATE TABLE IF NOT EXISTS events (
|
||||
id TEXT PRIMARY KEY,
|
||||
type TEXT NOT NULL DEFAULT '',
|
||||
at INTEGER NOT NULL DEFAULT 0,
|
||||
note TEXT NOT NULL DEFAULT '',
|
||||
photo_id TEXT NOT NULL DEFAULT '',
|
||||
weight REAL NOT NULL DEFAULT 0,
|
||||
updated INTEGER NOT NULL DEFAULT 0,
|
||||
deleted INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS config (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
name TEXT NOT NULL DEFAULT '',
|
||||
birthday TEXT NOT NULL DEFAULT '',
|
||||
updated INTEGER NOT NULL DEFAULT 0
|
||||
);`
|
||||
if _, err := db.Exec(schema); err != nil {
|
||||
db.Close()
|
||||
return nil, err
|
||||
}
|
||||
return db, nil
|
||||
}
|
||||
|
||||
// migrateJSON imports a pre-SQLite events.json / config.json sitting in dataDir
|
||||
// into an otherwise-empty database, then renames each file to *.imported so the
|
||||
// import runs exactly once. It is a no-op when the DB already holds data or the
|
||||
// legacy files are absent.
|
||||
func migrateJSON(db *sql.DB, dataDir string) error {
|
||||
if err := importEvents(db, filepath.Join(dataDir, "events.json")); err != nil {
|
||||
return err
|
||||
}
|
||||
return importConfig(db, filepath.Join(dataDir, "config.json"))
|
||||
}
|
||||
|
||||
func importEvents(db *sql.DB, path string) error {
|
||||
var n int
|
||||
if err := db.QueryRow(`SELECT COUNT(*) FROM events`).Scan(&n); err != nil {
|
||||
return err
|
||||
}
|
||||
if n > 0 {
|
||||
return nil // DB already has data; never clobber it
|
||||
}
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
var evs []Event
|
||||
if err := json.NewDecoder(f).Decode(&evs); err != nil {
|
||||
if errors.Is(err, io.EOF) {
|
||||
dec := json.NewDecoder(f)
|
||||
err = dec.Decode(&evs)
|
||||
f.Close()
|
||||
if err != nil && !errors.Is(err, io.EOF) {
|
||||
return err
|
||||
}
|
||||
store := newStore(db)
|
||||
if _, err := store.sync(evs); err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("migrated %d events from %s", len(evs), path)
|
||||
return os.Rename(path, path+".imported")
|
||||
}
|
||||
|
||||
func importConfig(db *sql.DB, path string) error {
|
||||
var n int
|
||||
if err := db.QueryRow(`SELECT COUNT(*) FROM config`).Scan(&n); err != nil {
|
||||
return err
|
||||
}
|
||||
if n > 0 {
|
||||
return nil
|
||||
}
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
for _, e := range evs {
|
||||
s.data[e.ID] = e
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Caller must hold s.mu.
|
||||
func (s *Store) saveLocked() error {
|
||||
if err := os.MkdirAll(filepath.Dir(s.path), 0o755); err != nil {
|
||||
var c Config
|
||||
dec := json.NewDecoder(f)
|
||||
err = dec.Decode(&c)
|
||||
f.Close()
|
||||
if err != nil && !errors.Is(err, io.EOF) {
|
||||
return err
|
||||
}
|
||||
tmp := s.path + ".tmp"
|
||||
f, err := os.Create(tmp)
|
||||
if err != nil {
|
||||
if c.UpdatedAt == 0 {
|
||||
// Nothing meaningful to import; leave config empty.
|
||||
return os.Rename(path, path+".imported")
|
||||
}
|
||||
if _, err := newConfigStore(db).merge(c); err != nil {
|
||||
return err
|
||||
}
|
||||
evs := make([]Event, 0, len(s.data))
|
||||
for _, e := range s.data {
|
||||
evs = append(evs, e)
|
||||
}
|
||||
enc := json.NewEncoder(f)
|
||||
enc.SetIndent("", " ")
|
||||
if err := enc.Encode(evs); err != nil {
|
||||
f.Close()
|
||||
os.Remove(tmp)
|
||||
return err
|
||||
}
|
||||
if err := f.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp, s.path)
|
||||
}
|
||||
|
||||
// sync merges client events into the store using last-write-wins by UpdatedAt,
|
||||
// then returns the full merged set.
|
||||
func (s *Store) sync(client []Event) ([]Event, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
for _, ce := range client {
|
||||
if ce.ID == "" {
|
||||
continue
|
||||
}
|
||||
existing, ok := s.data[ce.ID]
|
||||
if !ok || ce.UpdatedAt > existing.UpdatedAt {
|
||||
s.data[ce.ID] = ce
|
||||
}
|
||||
}
|
||||
if err := s.saveLocked(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]Event, 0, len(s.data))
|
||||
for _, e := range s.data {
|
||||
out = append(out, e)
|
||||
}
|
||||
return out, nil
|
||||
log.Printf("migrated config from %s", path)
|
||||
return os.Rename(path, path+".imported")
|
||||
}
|
||||
|
||||
type syncRequest struct {
|
||||
@@ -213,19 +290,23 @@ func (c cacheControlFS) Open(name string) (http.File, error) { return c.root.Ope
|
||||
|
||||
func main() {
|
||||
addr := flag.String("addr", ":8080", "listen address (e.g. :8080 or 0.0.0.0:8080)")
|
||||
dataPath := flag.String("data", "events.json", "path to events JSON file")
|
||||
dataPath := flag.String("data", "puppy.db", "path to SQLite database file")
|
||||
staticDir := flag.String("static", "", "directory of static files to serve")
|
||||
flag.Parse()
|
||||
|
||||
store, err := newStore(*dataPath)
|
||||
db, err := openDB(*dataPath)
|
||||
if err != nil {
|
||||
log.Fatalf("load store: %v", err)
|
||||
log.Fatalf("open db: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
// One-time import of any pre-SQLite JSON data sitting next to the DB.
|
||||
if err := migrateJSON(db, filepath.Dir(*dataPath)); err != nil {
|
||||
log.Fatalf("migrate json: %v", err)
|
||||
}
|
||||
|
||||
configStore, err := newConfigStore(filepath.Join(filepath.Dir(*dataPath), "config.json"))
|
||||
if err != nil {
|
||||
log.Fatalf("load config: %v", err)
|
||||
}
|
||||
store := newStore(db)
|
||||
configStore := newConfigStore(db)
|
||||
|
||||
photosDir := filepath.Join(filepath.Dir(*dataPath), "photos")
|
||||
if err := os.MkdirAll(photosDir, 0o755); err != nil {
|
||||
|
||||
Reference in New Issue
Block a user