keeporg/internal/server/sync.go

package server

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"

	"cc.tiredsysadmin.keeporg/internal/blob"
	"cc.tiredsysadmin.keeporg/internal/kdbx"
	"cc.tiredsysadmin.keeporg/internal/store"
)

// dbView is what a client learns about a database it has access to.
type dbView struct {
	ID            string         `json:"id"`
	Name          string         `json:"name"`
	OwnerGroupID  string         `json:"owner_group_id"`
	Perm          store.Perm     `json:"perm"`
	MergeStrategy store.Strategy `json:"merge_strategy"`
	Version       int64          `json:"version"`
	SHA256        string         `json:"sha256"`
	Size          int64          `json:"size"`
	UpdatedAt     time.Time      `json:"updated_at"`
	UpdatedBy     string         `json:"updated_by,omitempty"`
}

func toView(db *store.Database, p store.Perm) dbView {
	return dbView{ID: db.ID, Name: db.Name, OwnerGroupID: db.OwnerGroupID, Perm: p,
		MergeStrategy: db.MergeStrategy, Version: db.Version, SHA256: db.SHA256,
		Size: db.Size, UpdatedAt: db.UpdatedAt, UpdatedBy: db.UpdatedBy}
}

func (s *Server) handleListDatabases(w http.ResponseWriter, r *http.Request, u *store.User) {
	out := []dbView{}
	s.st.Read(func(st *store.State) {
		for _, a := range st.Assignments(u.ID) {
			out = append(out, toView(a.DB, a.Perm))
		}
	})
	writeJSON(w, http.StatusOK, map[string]any{"databases": out})
}

// handleGetBlob serves the current version of a database, or a historical one
// when ?version= is given.
func (s *Server) handleGetBlob(w http.ResponseWriter, r *http.Request, u *store.User) {
	id := r.PathValue("id")
	var (
		db   store.Database
		perm store.Perm
	)
	s.st.Read(func(st *store.State) {
		perm = st.DatabasePerm(u.ID, id)
		if d := st.Databases[id]; d != nil {
			db = *d
		}
	})
	if !perm.AtLeast(store.PermRead) {
		fail(w, http.StatusNotFound, "not_found", "no such database")
		return
	}
	version := db.Version
	if q := r.URL.Query().Get("version"); q != "" {
		v, err := strconv.ParseInt(q, 10, 64)
		if err != nil {
			fail(w, http.StatusBadRequest, "bad_request", "version must be an integer")
			return
		}
		version = v
	}
	if version == 0 {
		// A database that has never been written has no bytes yet.
		w.Header().Set("ETag", `"0"`)
		w.Header().Set("Content-Type", "application/octet-stream")
		w.WriteHeader(http.StatusNoContent)
		return
	}
	data, err := s.blobs.Get(id, version)
	if err != nil {
		fail(w, http.StatusNotFound, "not_found", "that version is no longer stored")
		return
	}
	w.Header().Set("ETag", etag(version))
	w.Header().Set("X-KeepOrg-SHA256", blob.SHA256(data))
	w.Header().Set("X-KeepOrg-Version", strconv.FormatInt(version, 10))
	w.Header().Set("Content-Type", "application/octet-stream")
	w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%q", db.Name+".kdbx"))
	w.Header().Set("Content-Length", strconv.Itoa(len(data)))
	_, _ = w.Write(data)
}

// handlePutBlob stores a new version. The client states the version it edited
// in If-Match; the stored merge strategy decides what happens if that is no
// longer current.
func (s *Server) handlePutBlob(w http.ResponseWriter, r *http.Request, u *store.User) {
	id := r.PathValue("id")
	var (
		perm     store.Perm
		current  int64
		strategy store.Strategy
		exists   bool
	)
	s.st.Read(func(st *store.State) {
		perm = st.DatabasePerm(u.ID, id)
		if d := st.Databases[id]; d != nil {
			exists, current, strategy = true, d.Version, d.MergeStrategy
		}
	})
	if !exists || !perm.AtLeast(store.PermRead) {
		fail(w, http.StatusNotFound, "not_found", "no such database")
		return
	}
	if !perm.AtLeast(store.PermWrite) {
		fail(w, http.StatusForbidden, "forbidden", "you have read-only access to this database")
		return
	}
	v := decidePush(current, r.Header.Get("If-Match"), strategy)
	// A merge vault answers a stale push by folding the two copies together
	// instead of refusing it, so the body is read either way.
	merging := !v.Accept && v.Code == "conflict" && strategy == store.StrategyMerge
	if !v.Accept && !merging {
		w.Header().Set("ETag", etag(current))
		w.Header().Set("X-KeepOrg-Strategy", string(strategy))
		writeJSON(w, v.Status, map[string]any{
			"code": v.Code, "error": v.Message,
			"current_version": current, "merge_strategy": strategy,
		})
		return
	}
	data, err := io.ReadAll(http.MaxBytesReader(w, r.Body, s.cfg.MaxDBSize))
	if err != nil {
		fail(w, http.StatusRequestEntityTooLarge, "too_large",
			fmt.Sprintf("database exceeds the %d byte limit", s.cfg.MaxDBSize))
		return
	}
	if len(data) == 0 {
		fail(w, http.StatusBadRequest, "bad_request", "empty body")
		return
	}
	sum := blob.SHA256(data)
	if want := r.Header.Get("X-KeepOrg-SHA256"); want != "" && !strings.EqualFold(want, sum) {
		fail(w, http.StatusBadRequest, "checksum_mismatch", "body does not match the declared digest")
		return
	}

	// The server opens a merge vault, so a file it will one day have to
	// open must stay within what it can afford to: a header demanding a
	// billion KDF rounds is refused here, before it is stored and before
	// any merge tries it.
	if strategy == store.StrategyMerge {
		if err := kdbx.Inspect(data); err != nil {
			fail(w, http.StatusBadRequest, "invalid_kdbx",
				"this server merges this vault, and cannot accept this file: "+err.Error())
			return
		}
	}

	if merging {
		s.mergePush(w, r, u, id, current, data)
		return
	}

	var (
		next      int64
		unchanged bool
		watchers  []string
		name      string
	)
	write := false
	err = s.st.UpdateWith(func(st *store.State) error {
		write = false
		d := st.Databases[id]
		if d == nil {
			return errors.New("no such database")
		}
		// Re-check under the write lock: a concurrent push may have landed.
		if vv := decidePush(d.Version, r.Header.Get("If-Match"), d.MergeStrategy); !vv.Accept {
			return conflictError{current: d.Version, strategy: d.MergeStrategy, verdict: vv}
		}
		if d.SHA256 == sum {
			next, unchanged, name = d.Version, true, d.Name
			return nil
		}
		next, write = d.Version+1, true
		d.Version, d.SHA256, d.Size = next, sum, int64(len(data))
		d.UpdatedAt, d.UpdatedBy = time.Now(), u.Username
		name = d.Name
		watchers = st.Watchers(id)
		return nil
	}, func(ctx context.Context, tx *sql.Tx) error {
		if !write {
			return nil
		}
		return s.blobs.PutTx(ctx, tx, id, next, data)
	})
	var ce conflictError
	switch {
	case errors.As(err, &ce):
		w.Header().Set("ETag", etag(ce.current))
		w.Header().Set("X-KeepOrg-Strategy", string(ce.strategy))
		writeJSON(w, ce.verdict.Status, map[string]any{
			"code": ce.verdict.Code, "error": ce.verdict.Message,
			"current_version": ce.current, "merge_strategy": ce.strategy,
		})
		return
	case err != nil:
		s.log.Printf("push %s: %v", id, err)
		fail(w, http.StatusInternalServerError, "internal", "the push could not be stored; try again")
		return
	}
	if !unchanged {
		_ = s.blobs.Prune(id)
		s.hub.Publish(watchers, Event{Type: EventUpdated, DatabaseID: id, Name: name,
			Version: next, SHA256: sum, By: u.Username})
	}
	w.Header().Set("ETag", etag(next))
	writeJSON(w, http.StatusOK, map[string]any{
		"version": next, "sha256": sum, "unchanged": unchanged,
	})
}

// handleEvents streams database updates to one client over SSE. SSE is enough
// here because the channel only ever carries server-to-client notifications.
func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request, u *store.User) {
	flusher, ok := w.(http.Flusher)
	if !ok {
		fail(w, http.StatusInternalServerError, "internal", "streaming unsupported")
		return
	}
	token := bearer(r)
	id, ch := s.hub.Subscribe(u.ID)
	defer s.hub.Unsubscribe(id)

	h := w.Header()
	h.Set("Content-Type", "text/event-stream")
	h.Set("Cache-Control", "no-store")
	h.Set("Connection", "keep-alive")
	h.Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)
	fmt.Fprintf(w, "retry: 3000\n\n")
	flusher.Flush()

	keepalive := time.NewTicker(25 * time.Second)
	defer keepalive.Stop()
	for {
		select {
		case <-r.Context().Done():
			return
		case ev, open := <-ch:
			if !open {
				return
			}
			fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, ev.encode())
			flusher.Flush()
		case <-keepalive.C:
			// A stream outlives the request that authenticated it: end it
			// once the session is revoked or the account disabled.
			if !s.sessionAlive(token) {
				return
			}
			fmt.Fprint(w, ": keepalive\n\n")
			flusher.Flush()
		}
	}
}

// ---- the push decision, kept pure ----

// verdict is the outcome of comparing a push against the stored version.
type verdict struct {
	Accept  bool
	Status  int
	Code    string
	Message string
}

type conflictError struct {
	current  int64
	strategy store.Strategy
	verdict  verdict
}

func (e conflictError) Error() string { return e.verdict.Message }

// decidePush is the whole merge policy. The client sends the version it based
// its edit on in If-Match; If-Match: * means "I know I am stale and I intend
// to win", which only a client-wins database allows.
func decidePush(current int64, ifMatch string, strategy store.Strategy) verdict {
	ifMatch = strings.TrimSpace(ifMatch)
	switch {
	case ifMatch == "":
		return verdict{Status: http.StatusPreconditionRequired, Code: "precondition_required",
			Message: "If-Match must carry the version this edit was based on"}
	case ifMatch == "*":
		if strategy == store.StrategyClientWins {
			return verdict{Accept: true}
		}
		return verdict{Status: http.StatusConflict, Code: "force_not_allowed",
			Message: "this database does not allow a client to overwrite the server copy"}
	}
	v, err := strconv.ParseInt(strings.Trim(ifMatch, `"`), 10, 64)
	if err != nil {
		return verdict{Status: http.StatusBadRequest, Code: "bad_request",
			Message: "If-Match must be a version number or *"}
	}
	if v == current {
		return verdict{Accept: true}
	}
	return verdict{Status: http.StatusConflict, Code: "conflict",
		Message: fmt.Sprintf("the server is at version %d, this edit was based on %d", current, v)}
}

func etag(v int64) string { return `"` + strconv.FormatInt(v, 10) + `"` }