keeporg/internal/client/api.go

// Package client is the shared client engine: one API wrapper and one
// synchronization agent, used by the desktop daemon and the admin CLI.
package client

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"
)

// maxPayload bounds one database download.
const maxPayload = 256 << 20

// API talks to one KeepOrg server as one user.
type API struct {
	BaseURL string
	Token   string
	HTTP    *http.Client
}

// NewAPI returns a client for a server base URL.
func NewAPI(baseURL, token string) *API {
	return &API{
		BaseURL: strings.TrimRight(baseURL, "/"),
		Token:   token,
		HTTP:    &http.Client{Timeout: 60 * time.Second},
	}
}

// Error is a structured failure returned by the server.
type Error struct {
	Status  int    `json:"-"`
	Code    string `json:"code"`
	Message string `json:"error"`
	// Conflict detail, present when Code is "conflict".
	CurrentVersion int64  `json:"current_version"`
	MergeStrategy  string `json:"merge_strategy"`
}

func (e *Error) Error() string {
	if e.Message == "" {
		return fmt.Sprintf("server returned %d", e.Status)
	}
	return e.Message
}

// IsConflict reports whether err is a rejected push.
func IsConflict(err error) bool {
	var e *Error
	return errors.As(err, &e) && (e.Code == "conflict" || e.Code == "force_not_allowed")
}

// Code returns the server's error code for err, or "".
func Code(err error) string {
	var e *Error
	if errors.As(err, &e) {
		return e.Code
	}
	return ""
}

// Database is one database as the client sees it.
type Database struct {
	ID            string    `json:"id"`
	Name          string    `json:"name"`
	OwnerGroupID  string    `json:"owner_group_id"`
	Perm          string    `json:"perm"`
	MergeStrategy string    `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"`
}

// CanWrite reports whether the user may push to this database.
func (d Database) CanWrite() bool { return d.Perm == "write" || d.Perm == "admin" }

// Session is the result of a successful login.
type Session struct {
	Token     string    `json:"token"`
	ExpiresAt time.Time `json:"expires_at"`
	User      struct {
		ID          string `json:"id"`
		Username    string `json:"username"`
		ServerAdmin bool   `json:"server_admin"`
		TOTPEnabled bool   `json:"totp_enabled"`
	} `json:"user"`
}

// DeviceStart is a pending browser-approved login.
type DeviceStart struct {
	DeviceCode      string `json:"device_code"`
	UserCode        string `json:"user_code"`
	VerificationURL string `json:"verification_url"`
	Interval        int    `json:"interval"`
	ExpiresIn       int    `json:"expires_in"`
}

// PushResult reports what the server did with an accepted push.
type PushResult struct {
	Version   int64  `json:"version"`
	SHA256    string `json:"sha256"`
	Unchanged bool   `json:"unchanged"`
	// Merged is set when the server folded this push into another edit; the
	// stored version is then neither copy and must be pulled.
	Merged       bool   `json:"merged"`
	MergeSummary string `json:"merge_summary"`
}

// Login exchanges a password and optional second factor for a session.
func (a *API) Login(ctx context.Context, user, password, totp, label string) (*Session, error) {
	var out Session
	err := a.do(ctx, http.MethodPost, "/api/v1/auth/login", map[string]string{
		"username": user, "password": password, "totp": totp, "label": label}, &out)
	if err != nil {
		return nil, err
	}
	return &out, nil
}

// DeviceStart opens a browser-approved login, used for OIDC and for devices
// without a comfortable keyboard.
func (a *API) DeviceStart(ctx context.Context, label string) (*DeviceStart, error) {
	var out DeviceStart
	if err := a.do(ctx, http.MethodPost, "/api/v1/auth/device/start",
		map[string]string{"label": label}, &out); err != nil {
		return nil, err
	}
	return &out, nil
}

// DevicePoll checks whether a browser has approved the login yet. It returns
// a nil session while the request is still pending.
func (a *API) DevicePoll(ctx context.Context, deviceCode string) (*Session, error) {
	var out Session
	err := a.do(ctx, http.MethodPost, "/api/v1/auth/device/poll",
		map[string]string{"device_code": deviceCode}, &out)
	var e *Error
	switch {
	case errors.As(err, &e) && e.Status == http.StatusAccepted:
		return nil, nil
	case err != nil:
		return nil, err
	}
	return &out, nil
}

// Logout revokes this client's session.
func (a *API) Logout(ctx context.Context) error {
	return a.do(ctx, http.MethodPost, "/api/v1/auth/logout", nil, nil)
}

// Databases lists everything assigned to the user.
func (a *API) Databases(ctx context.Context) ([]Database, error) {
	var out struct {
		Databases []Database `json:"databases"`
	}
	if err := a.do(ctx, http.MethodGet, "/api/v1/databases", nil, &out); err != nil {
		return nil, err
	}
	return out.Databases, nil
}

// Pull downloads the current bytes of a database along with their version.
// A database that has never been written returns no bytes and version 0.
func (a *API) Pull(ctx context.Context, dbID string) ([]byte, int64, error) {
	req, err := a.request(ctx, http.MethodGet, "/api/v1/databases/"+url.PathEscape(dbID)+"/blob", nil)
	if err != nil {
		return nil, 0, err
	}
	resp, err := a.HTTP.Do(req)
	if err != nil {
		return nil, 0, err
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNoContent {
		return nil, 0, nil
	}
	if resp.StatusCode != http.StatusOK {
		return nil, 0, decodeError(resp)
	}
	// The server's own limit is tens of megabytes; a server that sends more
	// is not one to trust with memory.
	data, err := io.ReadAll(io.LimitReader(resp.Body, maxPayload+1))
	if err != nil {
		return nil, 0, err
	}
	if len(data) > maxPayload {
		return nil, 0, fmt.Errorf("the server sent more than %d MiB for one database", maxPayload>>20)
	}
	v, _ := strconv.ParseInt(strings.Trim(resp.Header.Get("ETag"), `"`), 10, 64)
	return data, v, nil
}

// Push uploads new bytes. baseVersion is the version the edit was based on;
// force is only accepted by a client-wins database.
func (a *API) Push(ctx context.Context, dbID string, baseVersion int64, data []byte, force bool) (*PushResult, error) {
	req, err := a.request(ctx, http.MethodPut,
		"/api/v1/databases/"+url.PathEscape(dbID)+"/blob", bytes.NewReader(data))
	if err != nil {
		return nil, err
	}
	if force {
		req.Header.Set("If-Match", "*")
	} else {
		req.Header.Set("If-Match", strconv.FormatInt(baseVersion, 10))
	}
	req.Header.Set("Content-Type", "application/octet-stream")
	resp, err := a.HTTP.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return nil, decodeError(resp)
	}
	var out PushResult
	if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&out); err != nil {
		return nil, err
	}
	return &out, nil
}

// Event is a push notification from the server.
type Event struct {
	Type       string `json:"type"`
	DatabaseID string `json:"database_id"`
	Name       string `json:"name"`
	Version    int64  `json:"version"`
	SHA256     string `json:"sha256"`
	By         string `json:"by"`
}

// Events streams server notifications until ctx is done or the stream breaks.
// The caller reconnects; losing events is safe because every notification
// only ever triggers a full reconcile.
func (a *API) Events(ctx context.Context, onEvent func(Event)) error {
	// The token goes in the header like everywhere else. In the URL it
	// would land in proxy logs, and in every error message Go makes about
	// a failed request, which this client logs.
	req, err := a.request(ctx, http.MethodGet, "/api/v1/events", nil)
	if err != nil {
		return err
	}
	req.Header.Set("Accept", "text/event-stream")
	// A streaming response must not be cut off by the client timeout.
	stream := &http.Client{Transport: a.transport()}
	resp, err := stream.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return decodeError(resp)
	}
	dec := newSSEDecoder(resp.Body)
	for {
		ev, err := dec.next()
		if err != nil {
			return err
		}
		if ev == nil {
			continue
		}
		onEvent(*ev)
	}
}

func (a *API) transport() http.RoundTripper {
	if a.HTTP != nil && a.HTTP.Transport != nil {
		return a.HTTP.Transport
	}
	return http.DefaultTransport
}

func (a *API) request(ctx context.Context, method, path string, body io.Reader) (*http.Request, error) {
	req, err := http.NewRequestWithContext(ctx, method, a.BaseURL+path, body)
	if err != nil {
		return nil, err
	}
	if a.Token != "" {
		req.Header.Set("Authorization", "Bearer "+a.Token)
	}
	return req, nil
}

// do sends a JSON request and decodes a JSON response. A non-2xx status
// always becomes an *Error.
func (a *API) do(ctx context.Context, method, path string, in, out any) error {
	var body io.Reader
	if in != nil {
		b, err := json.Marshal(in)
		if err != nil {
			return err
		}
		body = bytes.NewReader(b)
	}
	req, err := a.request(ctx, method, path, body)
	if err != nil {
		return err
	}
	if in != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	resp, err := a.HTTP.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode > 299 {
		return decodeError(resp)
	}
	if resp.StatusCode == http.StatusAccepted {
		return &Error{Status: resp.StatusCode, Code: "pending"}
	}
	if out == nil {
		return nil
	}
	return json.NewDecoder(io.LimitReader(resp.Body, 8<<20)).Decode(out)
}

func decodeError(resp *http.Response) error {
	e := &Error{Status: resp.StatusCode}
	b, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
	if len(b) > 0 && json.Unmarshal(b, e) == nil && e.Message != "" {
		return e
	}
	e.Message = strings.TrimSpace(string(b))
	if e.Message == "" {
		e.Message = resp.Status
	}
	return e
}