Take in an admin's edits of the state files while running (closes #68)

smallwebwaf watches SWWAF_STATE_DIR with fsnotify and takes in an edit of
bans.json, clients.json or lookups.json as soon as it is saved, in place
of what it held. It tells its own writes from an admin's by the SHA-256
of what it last read or wrote, and each write takes in an edit made since
first. An edit that does not parse is renamed to <name>.bad at the
file's next write, which writes the file again from memory and logs the
file and where the error is. README.md says how to add and lift a ban.

Judgement call: a broken edit is set aside at the file's next write, not
when seen, since an editor's file can be read half written.
Judgement call: a state file that cannot be read is not written over.

Model: opus-5-5

internal/state/state.go

@@ -1,14 +1,16 @@
 // Package state keeps smallwebwaf's state in JSON files in
 // SWWAF_STATE_DIR, as the "Persistent state" section of SPEC.md describes:
 // bans.json holds the bans, clients.json each client's counters and
-// history, and lookups.json GeoJS's answers. Load reads them at start, and
-// Run and WriteAll write them, each from a snapshot its part takes under
-// its own lock, so that no request waits on the disk.
+// history, and lookups.json GeoJS's answers. Load reads them at start,
+// Watch takes in an admin's edit of one while smallwebwaf runs, and Run
+// and WriteAll write them. Each part takes a snapshot, or what a file
+// holds, under its own lock, so that no request waits on the disk.
 package state
 
 import (
 	"bytes"
 	"context"
+	"crypto/sha256"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -17,8 +19,10 @@ import (
 	"net/netip"
 	"os"
 	"path/filepath"
+	"sync"
 	"time"
 
+	"github.com/fsnotify/fsnotify"
 	"sneak.berlin/go/smallwebwaf/internal/bans"
 	"sneak.berlin/go/smallwebwaf/internal/lookup"
 	"sneak.berlin/go/smallwebwaf/internal/ratelimit"
@@ -60,13 +64,23 @@ type Params struct {
 	// Now tells the time by which the counters' buckets run out, normally
 	// time.Now in UTC.
 	Now func() time.Time
-	// ProcessLog receives what was read, and the writes that fail.
+	// ProcessLog receives what was read and taken in, the edits set aside,
+	// and the writes that fail.
 	ProcessLog *slog.Logger
 }
 
 // Files are the state files of a running smallwebwaf.
 type Files struct {
 	params Params
+
+	// mu is held while a file is read for an edit, and while it is
+	// written, so that Watch and the writes take turns. No request takes
+	// it.
+	mu sync.Mutex
+	// sums are the SHA-256 sums of what each file held, by name, when
+	// smallwebwaf last read or wrote it. A file that holds anything else
+	// has been edited since.
+	sums map[string][sha256.Size]byte
 }
 
 // bansFile is bans.json, indented for an admin to read and edit.
@@ -117,41 +131,28 @@ func Load(params Params) (*Files, error) {
 		return nil, fmt.Errorf("SWWAF_STATE_DIR cannot be written: %w", err)
 	}
 
-	var (
-		bansIn    bansFile
-		clientsIn clientsFile
-		lookupsIn lookupsFile
-	)
+	f := &Files{params: params, sums: map[string][sha256.Size]byte{}}
 
-	err = errors.Join(
-		read(params.Dir, bansJSON, &bansIn),
-		read(params.Dir, clientsJSON, &clientsIn),
-		read(params.Dir, lookupsJSON, &lookupsIn),
-	)
+	bansRead, bansErr := f.read(bansJSON)
+	clientsRead, clientsErr := f.read(clientsJSON)
+	lookupsRead, lookupsErr := f.read(lookupsJSON)
+
+	err = errors.Join(bansErr, clientsErr, lookupsErr)
 	if err != nil {
 		return nil, err
 	}
 
-	held := make([]bans.Ban, 0, len(bansIn.Bans))
-	for _, entry := range bansIn.Bans {
-		held = append(held, entry.ban())
-	}
-
-	params.Ledger.Load(held)
-	params.Limiter.Load(clientsIn.Clients, params.Now())
-	params.GeoJS.Load(lookupsIn.Lookups)
-
 	params.ProcessLog.Info("read the state files", "directory", params.Dir,
-		"bans", len(bansIn.Bans), "clients", len(clientsIn.Clients),
-		"lookups", len(lookupsIn.Lookups))
+		"bans", bansRead, "clients", clientsRead, "lookups", lookupsRead)
 
-	return &Files{params: params}, nil
+	return f, nil
 }
 
 // Run writes bans.json WriteDelay after a ban is made, with every ban
 // made in between, and every file every CounterInterval, until ctx is
 // done. A write that fails is logged, and the file is written again at
-// its next write.
+// its next write. Each write takes in an admin's edit of its file first,
+// as writeFile describes.
 func (f *Files) Run(ctx context.Context) {
 	interval := time.NewTicker(f.params.CounterInterval)
 	defer interval.Stop()
@@ -169,7 +170,7 @@ func (f *Files) Run(ctx context.Context) {
 		case <-bansDue:
 			bansDue = nil
 
-			f.logFailure(f.writeBans())
+			f.logFailure(f.writeFile(bansJSON))
 		case <-interval.C:
 			f.logFailure(f.WriteAll())
 		}
@@ -179,7 +180,50 @@ func (f *Files) Run(ctx context.Context) {
 // WriteAll writes every state file, as smallwebwaf stops. A file that
 // fails does not keep the others from being written.
 func (f *Files) WriteAll() error {
-	return errors.Join(f.writeBans(), f.writeClients(), f.writeLookups())
+	return errors.Join(f.writeFile(bansJSON), f.writeFile(clientsJSON),
+		f.writeFile(lookupsJSON))
+}
+
+// Watch watches Dir until ctx is done, and takes in an admin's edit of a
+// state file as soon as it is saved: what the file holds replaces what
+// smallwebwaf held for it. An edit that does not parse is left for the
+// file's next write, which sets it aside, since a file can be read while
+// an editor is still writing it. If Dir cannot be watched, that is
+// logged, and an edit is taken in only before its file is written.
+func (f *Files) Watch(ctx context.Context) {
+	watcher, err := fsnotify.NewWatcher()
+	if err == nil {
+		defer func() {
+			_ = watcher.Close()
+		}()
+
+		err = watcher.Add(f.params.Dir)
+	}
+
+	if err != nil {
+		f.params.ProcessLog.Error("cannot watch the state files for edits",
+			"error", err.Error())
+
+		return
+	}
+
+	f.params.ProcessLog.Info("watching the state files for edits",
+		"directory", f.params.Dir)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case event := <-watcher.Events:
+			switch name := filepath.Base(event.Name); name {
+			case bansJSON, clientsJSON, lookupsJSON:
+				f.takeInEdit(name)
+			}
+		case err = <-watcher.Errors:
+			f.params.ProcessLog.Warn("watching the state files failed",
+				"error", err.Error())
+		}
+	}
 }
 
 // logFailure logs a write that failed.
@@ -190,41 +234,177 @@ func (f *Files) logFailure(err error) {
 	}
 }
 
-// writeBans writes bans.json.
-func (f *Files) writeBans() error {
-	held := f.params.Ledger.Snapshot()
+// takeInEdit takes in an edit of the state file name and logs it, if the
+// file has changed since smallwebwaf last read or wrote it and parses. A
+// file that cannot be read or does not parse is left for its next write.
+func (f *Files) takeInEdit(name string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 
-	file := bansFile{Version: version, Bans: make([]banEntry, 0, len(held))}
-	for _, ban := range held {
-		file.Bans = append(file.Bans, newBanEntry(ban))
+	data, changed, err := f.readChanged(name)
+	if err != nil || !changed {
+		return
 	}
 
-	data, err := json.MarshalIndent(file, "", "    ")
+	_, err = f.takeIn(name, data)
 	if err != nil {
-		return fmt.Errorf("encode %s: %w", bansJSON, err)
+		return
 	}
 
-	return write(f.params.Dir, bansJSON, append(data, '\n'))
+	f.params.ProcessLog.Info("took in an edit of a state file",
+		"file", filepath.Join(f.params.Dir, name))
 }
 
-// writeClients writes clients.json.
-func (f *Files) writeClients() error {
-	data, err := encodeOnePerLine("clients", f.params.Limiter.Snapshot())
-	if err != nil {
-		return fmt.Errorf("encode %s: %w", clientsJSON, err)
+// read takes in the state file name, as at start, and returns how many
+// entries it holds. A missing file holds none, and is not read.
+func (f *Files) read(name string) (int, error) {
+	data, changed, err := f.readChanged(name)
+	if err != nil || !changed {
+		return 0, err
 	}
 
-	return write(f.params.Dir, clientsJSON, data)
+	return f.takeIn(name, data)
 }
 
-// writeLookups writes lookups.json.
-func (f *Files) writeLookups() error {
-	data, err := encodeOnePerLine("lookups", f.params.GeoJS.Snapshot())
-	if err != nil {
-		return fmt.Errorf("encode %s: %w", lookupsJSON, err)
+// readChanged returns what the state file name holds, and whether that
+// has changed since smallwebwaf last read or wrote the file, as it has
+// for a file smallwebwaf never read or wrote. A missing file has not
+// changed: it is written again at its next write.
+func (f *Files) readChanged(name string) ([]byte, bool, error) {
+	path := filepath.Join(f.params.Dir, name)
+
+	data, err := os.ReadFile(path) //nolint:gosec // a state file, in SWWAF_STATE_DIR
+	if errors.Is(err, fs.ErrNotExist) {
+		return nil, false, nil
 	}
 
-	return write(f.params.Dir, lookupsJSON, data)
+	if err != nil {
+		return nil, false, err
+	}
+
+	return data, sha256.Sum256(data) != f.sums[name], nil
+}
+
+// takeIn parses data, what the state file name holds, puts it into the
+// part that keeps that state, in place of what the part held, and returns
+// how many entries the file holds. An error names the file and, where the
+// JSON decoder tells it, the line and column, or else the entry.
+func (f *Files) takeIn(name string, data []byte) (int, error) {
+	path := filepath.Join(f.params.Dir, name)
+
+	var entries int
+
+	switch name {
+	case bansJSON:
+		var file bansFile
+
+		err := parse(path, data, &file)
+		if err != nil {
+			return 0, err
+		}
+
+		held := make([]bans.Ban, 0, len(file.Bans))
+		for _, entry := range file.Bans {
+			held = append(held, entry.ban())
+		}
+
+		f.params.Ledger.Load(held)
+		entries = len(held)
+	case clientsJSON:
+		var file clientsFile
+
+		err := parse(path, data, &file)
+		if err != nil {
+			return 0, err
+		}
+
+		f.params.Limiter.Load(file.Clients, f.params.Now())
+		entries = len(file.Clients)
+	case lookupsJSON:
+		var file lookupsFile
+
+		err := parse(path, data, &file)
+		if err != nil {
+			return 0, err
+		}
+
+		f.params.GeoJS.Load(file.Lookups)
+		entries = len(file.Lookups)
+	}
+
+	f.sums[name] = sha256.Sum256(data)
+
+	return entries, nil
+}
+
+// writeFile writes the state file name from what smallwebwaf holds. An
+// edit made since smallwebwaf last read or wrote the file is taken in
+// first, so that it is not overwritten. An edit that does not parse is
+// renamed to name.bad, for the admin to mend, and logged with where in
+// the file the error is.
+func (f *Files) writeFile(name string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	path := filepath.Join(f.params.Dir, name)
+
+	data, changed, err := f.readChanged(name)
+	if err != nil {
+		return err
+	}
+
+	if changed {
+		_, err = f.takeIn(name, data)
+	}
+
+	if err != nil {
+		renameErr := os.Rename(path, path+".bad")
+		if renameErr != nil {
+			return errors.Join(err, renameErr)
+		}
+
+		f.params.ProcessLog.Error("set aside an edit of a state file that does not parse",
+			"file", path+".bad", "error", err.Error())
+	}
+
+	data, err = f.encode(name)
+	if err != nil {
+		return fmt.Errorf("encode %s: %w", name, err)
+	}
+
+	err = write(f.params.Dir, name, data)
+	if err != nil {
+		return err
+	}
+
+	f.sums[name] = sha256.Sum256(data)
+
+	return nil
+}
+
+// encode returns the state file name as smallwebwaf writes it, from a
+// snapshot of the part that keeps that state.
+func (f *Files) encode(name string) ([]byte, error) {
+	switch name {
+	case bansJSON:
+		held := f.params.Ledger.Snapshot()
+
+		file := bansFile{Version: version, Bans: make([]banEntry, 0, len(held))}
+		for _, ban := range held {
+			file.Bans = append(file.Bans, newBanEntry(ban))
+		}
+
+		data, err := json.MarshalIndent(file, "", "    ")
+		if err != nil {
+			return nil, err
+		}
+
+		return append(data, '\n'), nil
+	case clientsJSON:
+		return encodeOnePerLine("clients", f.params.Limiter.Snapshot())
+	default: // lookups.json
+		return encodeOnePerLine("lookups", f.params.GeoJS.Snapshot())
+	}
 }
 
 // newBanEntry returns ban as bans.json holds it.
@@ -377,28 +557,16 @@ func checkWritable(dir string) error {
 	return errors.Join(file.Close(), os.Remove(file.Name()))
 }
 
-// read reads the state file name in dir into file, a pointer to that
-// file's struct, and checks its entries. A missing file leaves file as it
-// is.
-func read(dir, name string, file stateFile) error {
-	path := filepath.Join(dir, name)
-
-	data, err := os.ReadFile(path) //nolint:gosec // a state file, in SWWAF_STATE_DIR
-	if errors.Is(err, fs.ErrNotExist) {
-		return nil
-	}
-
-	if err != nil {
-		return err
-	}
-
+// parse reads data, what the state file at path holds, into file, a
+// pointer to that file's struct, and checks its entries.
+func parse(path string, data []byte, file stateFile) error {
 	// The version is read first, so that a file of another version is
 	// refused for that, and not for an entry this version cannot read.
 	var header struct {
 		Version int `json:"version"`
 	}
 
-	err = json.Unmarshal(data, &header)
+	err := json.Unmarshal(data, &header)
 	if err == nil && header.Version != version {
 		err = fmt.Errorf("%w %d, where this smallwebwaf reads version %d",
 			errVersion, header.Version, version)