keeporg/internal/client/mirror.go

package client

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"
)

// stateFile holds the mirror's bookkeeping. It sits inside the mirror
// directory so moving the directory moves its history with it.
const stateFile = ".keeporg-state.json"

// fileState is what this client last agreed with the server about one
// database, and where it keeps it on disk.
type fileState struct {
	Name           string `json:"name"`
	File           string `json:"file"`
	Version        int64  `json:"version"`
	SHA256         string `json:"sha256"`
	ConflictMarker string `json:"conflict_marker,omitempty"`
}

// Status is a human-readable account of one mirrored database.
type Status struct {
	DatabaseID string    `json:"database_id"`
	Name       string    `json:"name"`
	File       string    `json:"file"`
	Perm       string    `json:"perm"`
	Strategy   string    `json:"merge_strategy"`
	Version    int64     `json:"version"`
	State      string    `json:"state"`
	Message    string    `json:"message,omitempty"`
	LastSync   time.Time `json:"last_sync"`
}

// Mirror keeps a directory of real .kdbx files in step with a server. Any
// KeePass application can open the files directly; the mirror notices a save
// and pushes it, and applies what other clients push.
//
// Every reconciliation holds m.mu, so there is exactly one writer to the
// directory and to the state file.
type Mirror struct {
	api  *API
	dir  string
	log  *log.Logger
	poll time.Duration

	// kick asks Run for a reconcile now. It holds at most one request, so
	// a burst of them costs one pass.
	kick chan struct{}

	mu     sync.Mutex
	st     map[string]*fileState
	status map[string]*Status
	// lastSync and lastErr describe the most recent SyncAll as a whole, which
	// a per-database status cannot: an unreachable server has no databases.
	lastSync time.Time
	lastErr  error
}

// NewMirror prepares dir and loads any previous state.
func NewMirror(api *API, dir string, lg *log.Logger, poll time.Duration) (*Mirror, error) {
	if err := os.MkdirAll(dir, 0o700); err != nil {
		return nil, err
	}
	if poll <= 0 {
		poll = 2 * time.Second
	}
	m := &Mirror{api: api, dir: dir, log: lg, poll: poll, kick: make(chan struct{}, 1),
		st: map[string]*fileState{}, status: map[string]*Status{}}
	b, err := os.ReadFile(filepath.Join(dir, stateFile))
	switch {
	case os.IsNotExist(err):
		return m, nil
	case err != nil:
		return nil, err
	}
	if err := json.Unmarshal(b, &m.st); err != nil {
		return nil, fmt.Errorf("mirror: parse %s: %w", stateFile, err)
	}
	return m, nil
}

// Dir is the mirror directory.
func (m *Mirror) Dir() string { return m.dir }

// Kick asks a running mirror to reconcile now instead of at the next rescan.
func (m *Mirror) Kick() {
	select {
	case m.kick <- struct{}{}:
	default:
	}
}

// Run reconciles until ctx is done, driven by server push events and by a
// local rescan for saves made by a KeePass application. It holds the mirror
// directory's lock throughout, so a second agent on the same directory fails
// to start instead of racing this one.
func (m *Mirror) Run(ctx context.Context) error {
	unlock, err := LockDir(m.dir)
	if err != nil {
		return err
	}
	defer unlock()

	go m.listen(ctx, m.Kick)

	if err := m.SyncAll(ctx); err != nil {
		m.log.Printf("sync: %v", err)
	}
	t := time.NewTicker(m.poll)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-t.C:
			if err := m.SyncAll(ctx); err != nil {
				m.log.Printf("sync: %v", err)
			}
		case <-m.kick:
			// Let a burst of events settle before reconciling.
			time.Sleep(150 * time.Millisecond)
			if err := m.SyncAll(ctx); err != nil {
				m.log.Printf("sync: %v", err)
			}
		}
	}
}

// listen keeps an event stream open, reconnecting with backoff. Losing the
// stream only delays a reconcile, it never loses data.
func (m *Mirror) listen(ctx context.Context, wake func()) {
	backoff := time.Second
	for ctx.Err() == nil {
		err := m.api.Events(ctx, func(ev Event) {
			backoff = time.Second
			m.log.Printf("event %s %s v%d by %s", ev.Type, ev.Name, ev.Version, ev.By)
			wake()
		})
		if ctx.Err() != nil {
			return
		}
		if err != nil && !errors.Is(err, io.EOF) {
			m.log.Printf("event stream: %v (retrying in %s)", err, backoff)
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(backoff):
		}
		if backoff < 60*time.Second {
			backoff *= 2
		}
	}
}

// SyncAll reconciles every assigned database once.
func (m *Mirror) SyncAll(ctx context.Context) error {
	dbs, err := m.api.Databases(ctx)
	if err != nil {
		m.mu.Lock()
		m.lastSync, m.lastErr = time.Now(), err
		m.mu.Unlock()
		return err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	defer func() { m.lastSync = time.Now() }()

	seen := map[string]bool{}
	var firstErr error
	for _, db := range dbs {
		seen[db.ID] = true
		if err := m.reconcile(ctx, db); err != nil {
			m.setStatus(db, "error", err.Error())
			m.log.Printf("%s: %v", db.Name, err)
			if firstErr == nil {
				firstErr = err
			}
		}
	}
	// A database that is no longer assigned stops being tracked. Its file is
	// left alone: deleting a user's local copy of a password database is not
	// this program's decision.
	for id := range m.st {
		if !seen[id] {
			delete(m.st, id)
			delete(m.status, id)
		}
	}
	if err := m.saveState(); err != nil && firstErr == nil {
		firstErr = err
	}
	// A failure of one database is in its own status; lastErr is for what
	// stopped the whole pass.
	m.lastErr = nil
	return firstErr
}

// LastSync reports when the last pass finished, and the error that stopped
// it from reaching the server, if one did.
func (m *Mirror) LastSync() (time.Time, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.lastSync, m.lastErr
}

// Status reports the current state of each mirrored database.
func (m *Mirror) Status() []Status {
	m.mu.Lock()
	defer m.mu.Unlock()
	out := make([]Status, 0, len(m.status))
	for _, s := range m.status {
		out = append(out, *s)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
	return out
}

// reconcile brings one database and its local file into agreement.
func (m *Mirror) reconcile(ctx context.Context, db Database) error {
	st := m.trackedState(db)
	path := filepath.Join(m.dir, st.File)

	local, exists, err := readFileIfExists(path)
	if err != nil {
		return err
	}
	localSHA := ""
	if exists {
		localSHA = digest(local)
	}
	sit := situation{
		LocalExists: exists, LocalSHA: localSHA,
		SyncedSHA: st.SHA256, SyncedVersion: st.Version,
		RemoteSHA: db.SHA256, RemoteVersion: db.Version,
		CanWrite: db.CanWrite(), Strategy: db.MergeStrategy,
		ConflictSeen: st.ConflictMarker == conflictMarker(db.Version, localSHA),
	}
	act := decide(sit)
	if act != actManual && st.ConflictMarker != "" {
		st.ConflictMarker = ""
	}

	switch act {
	case actNothing:
		if m.status[db.ID] == nil || m.status[db.ID].State == "" {
			m.setStatus(db, "ok", "")
		}
		return nil

	case actPull:
		return m.pull(ctx, db, st, path)

	case actPush, actPushForMerge:
		return m.push(ctx, db, st, path, local, false)

	case actForcePush:
		return m.push(ctx, db, st, path, local, true)

	case actServerWins, actRestore:
		copyPath, err := m.keepConflictCopy(path, local)
		if err != nil {
			return err
		}
		if err := m.pull(ctx, db, st, path); err != nil {
			return err
		}
		why := "the server copy was newer, your save was kept as " + filepath.Base(copyPath)
		if act == actRestore {
			why = "you have read-only access, your save was kept as " + filepath.Base(copyPath)
		}
		m.log.Printf("%s: %s", db.Name, why)
		m.setStatus(db, "conflict-resolved", why)
		return nil

	case actManual:
		remote, version, err := m.api.Pull(ctx, db.ID)
		if err != nil {
			return err
		}
		side := strings.TrimSuffix(path, ".kdbx") + fmt.Sprintf(".server-v%d.kdbx", version)
		if err := writeFileAtomic(side, remote); err != nil {
			return err
		}
		st.ConflictMarker = conflictMarker(db.Version, localSHA)
		why := fmt.Sprintf("both copies changed; the server copy is in %s, merge it yourself",
			filepath.Base(side))
		m.log.Printf("%s: %s", db.Name, why)
		m.setStatus(db, "conflict", why)
		return nil
	}
	return nil
}

func (m *Mirror) pull(ctx context.Context, db Database, st *fileState, path string) error {
	data, version, err := m.api.Pull(ctx, db.ID)
	if err != nil {
		return err
	}
	if version == 0 || len(data) == 0 {
		return nil // nothing stored yet
	}
	if err := writeFileAtomic(path, data); err != nil {
		return err
	}
	st.Version, st.SHA256 = version, digest(data)
	m.log.Printf("%s: pulled version %d into %s", db.Name, version, st.File)
	m.setStatus(db, "ok", "")
	return nil
}

func (m *Mirror) push(ctx context.Context, db Database, st *fileState, path string, data []byte, force bool) error {
	res, err := m.api.Push(ctx, db.ID, st.Version, data, force)
	switch {
	case err == nil:
		st.Version, st.SHA256 = res.Version, res.SHA256
		if res.Merged {
			// The server folded our save into someone else's. What it holds
			// now is neither copy, so take it.
			m.log.Printf("%s: server merged version %d (%s)", db.Name, res.Version, res.MergeSummary)
			if err := m.pull(ctx, db, st, path); err != nil {
				return err
			}
			m.setStatus(db, "merged", res.MergeSummary)
			return nil
		}
		if res.SHA256 != digest(data) {
			// Whatever the server stored differs from what we sent, so the
			// local file is no longer what the vault holds.
			return m.pull(ctx, db, st, path)
		}
		if !res.Unchanged {
			m.log.Printf("%s: pushed version %d", db.Name, res.Version)
		}
		m.setStatus(db, "ok", "")
		return nil

	case IsConflict(err):
		// The server moved while we were deciding. Re-read the policy from
		// the server's own answer and resolve with it.
		var e *Error
		errors.As(err, &e)
		strategy := e.MergeStrategy
		if strategy == "" {
			strategy = db.MergeStrategy
		}
		if strategy == "client-wins" && !force {
			return m.push(ctx, db, st, path, data, true)
		}
		if e.Code == "merge_failed" {
			// The server could not merge, most often because this vault's
			// own password was changed. Keep our copy and take the server's,
			// which is exactly what server-wins does.
			m.log.Printf("%s: the server could not merge (%s)", db.Name, e.Message)
		}
		copyPath, cerr := m.keepConflictCopy(path, data)
		if cerr != nil {
			return cerr
		}
		if perr := m.pull(ctx, db, st, path); perr != nil {
			return perr
		}
		why := "the server copy was newer, your save was kept as " + filepath.Base(copyPath)
		m.log.Printf("%s: %s", db.Name, why)
		m.setStatus(db, "conflict-resolved", why)
		return nil

	default:
		var e *Error
		if errors.As(err, &e) && e.Status == http.StatusForbidden {
			copyPath, cerr := m.keepConflictCopy(path, data)
			if cerr != nil {
				return cerr
			}
			if perr := m.pull(ctx, db, st, path); perr != nil {
				return perr
			}
			why := "you have read-only access, your save was kept as " + filepath.Base(copyPath)
			m.setStatus(db, "conflict-resolved", why)
			return nil
		}
		return err
	}
}

// trackedState returns the state for a database, following a rename made on
// the server.
func (m *Mirror) trackedState(db Database) *fileState {
	st := m.st[db.ID]
	if st == nil {
		st = &fileState{Name: db.Name, File: m.fileNameFor(db)}
		m.st[db.ID] = st
		return st
	}
	if st.File == "" {
		st.File = m.fileNameFor(db)
	}
	if st.Name != db.Name {
		want := m.fileNameFor(db)
		old := filepath.Join(m.dir, st.File)
		if _, err := os.Stat(old); err == nil && want != st.File {
			if err := os.Rename(old, filepath.Join(m.dir, want)); err != nil {
				m.log.Printf("%s: rename %s to %s: %v", db.Name, st.File, want, err)
				return st
			}
			m.log.Printf("renamed %s to %s", st.File, want)
		}
		st.Name, st.File = db.Name, want
	}
	return st
}

// fileNameFor turns a database name into a filename, keeping it unique
// within the mirror directory.
func (m *Mirror) fileNameFor(db Database) string {
	base := sanitize(db.Name)
	if base == "" {
		base = "database"
	}
	name := base + ".kdbx"
	for id, st := range m.st {
		if id != db.ID && st.File == name {
			return fmt.Sprintf("%s-%s.kdbx", base, shortID(db.ID))
		}
	}
	return name
}

// keepConflictCopy preserves bytes that are about to be overwritten.
func (m *Mirror) keepConflictCopy(path string, data []byte) (string, error) {
	if len(data) == 0 {
		return path, nil
	}
	copyPath := fmt.Sprintf("%s.conflict-%s.kdbx",
		strings.TrimSuffix(path, ".kdbx"), time.Now().UTC().Format("20060102-150405"))
	return copyPath, writeFileAtomic(copyPath, data)
}

func (m *Mirror) setStatus(db Database, state, msg string) {
	st := m.st[db.ID]
	s := &Status{DatabaseID: db.ID, Name: db.Name, Perm: db.Perm,
		Strategy: db.MergeStrategy, State: state, Message: msg, LastSync: time.Now()}
	if st != nil {
		s.File, s.Version = st.File, st.Version
	}
	m.status[db.ID] = s
}

func (m *Mirror) saveState() error {
	b, err := json.MarshalIndent(m.st, "", "  ")
	if err != nil {
		return err
	}
	return writeFileAtomic(filepath.Join(m.dir, stateFile), append(b, '\n'))
}

// ---- small helpers ----

func digest(b []byte) string {
	sum := sha256.Sum256(b)
	return hex.EncodeToString(sum[:])
}

func conflictMarker(remoteVersion int64, localSHA string) string {
	if localSHA == "" {
		return ""
	}
	return fmt.Sprintf("%d/%s", remoteVersion, localSHA)
}

func readFileIfExists(path string) ([]byte, bool, error) {
	b, err := os.ReadFile(path)
	switch {
	case os.IsNotExist(err):
		return nil, false, nil
	case err != nil:
		return nil, false, err
	}
	return b, true, nil
}

// writeFileAtomic replaces path in one step, so a KeePass application never
// reads a half-written database.
func writeFileAtomic(path string, data []byte) error {
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return err
	}
	tmp := path + ".keeporg-tmp"
	f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
	if err != nil {
		return err
	}
	if _, err := f.Write(data); err != nil {
		f.Close()
		os.Remove(tmp)
		return err
	}
	if err := f.Sync(); err != nil {
		f.Close()
		os.Remove(tmp)
		return err
	}
	if err := f.Close(); err != nil {
		os.Remove(tmp)
		return err
	}
	return os.Rename(tmp, path)
}

// sanitize keeps a database name usable as a filename on every platform.
func sanitize(name string) string {
	var b strings.Builder
	for _, r := range strings.TrimSpace(name) {
		switch {
		case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9',
			r == '-', r == '_', r == '.', r == ' ':
			b.WriteRune(r)
		default:
			b.WriteRune('-')
		}
	}
	return strings.Trim(strings.TrimSuffix(b.String(), ".kdbx"), " .-")
}

func shortID(id string) string {
	if len(id) > 6 {
		return id[:6]
	}
	return id
}