feat: add sync backend methods

2026-06-20 02:25:25 +08:00
parent 5c979174f1
commit ed69746332
5 changed files with 1083 additions and 12 deletions
@@ -0,0 +1,336 @@
+package sync
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"net/http"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+// DeviceTokenPath returns the path to the device_token file inside the vault.
+func DeviceTokenPath(vaultRoot string) string {
+	return filepath.Join(vaultRoot, ".verstak", "device_token.json")
+}
+
+// SaveDeviceToken writes the device token to a file with 0600 perms.
+func SaveDeviceToken(vaultRoot, token string) error {
+	path := DeviceTokenPath(vaultRoot)
+	dir := filepath.Dir(path)
+	if err := os.MkdirAll(dir, 0o750); err != nil {
+		return err
+	}
+	data := fmt.Sprintf(`{"device_token":%q}`, token)
+	return os.WriteFile(path, []byte(data), 0o600)
+}
+
+// LoadDeviceToken reads the device token from the vault.
+func LoadDeviceToken(vaultRoot string) string {
+	path := DeviceTokenPath(vaultRoot)
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return ""
+	}
+	var v struct {
+		DeviceToken string `json:"device_token"`
+	}
+	if err := json.Unmarshal(data, &v); err != nil {
+		return ""
+	}
+	return v.DeviceToken
+}
+
+// RemoveDeviceToken deletes the device token file.
+func RemoveDeviceToken(vaultRoot string) error {
+	path := DeviceTokenPath(vaultRoot)
+	return os.Remove(path)
+}
+
+// Client communicates with the Verstak Sync Server.
+type Client struct {
+	ServerURL   string
+	APIKey      string
+	DeviceToken string
+	DeviceID    string
+	VaultRoot   string
+	HTTP        *http.Client
+}
+
+// NewClient creates a sync client.
+func NewClient(serverURL, apiKey, deviceID, vaultRoot string) *Client {
+	return &Client{
+		ServerURL: serverURL,
+		APIKey:    apiKey,
+		DeviceID:  deviceID,
+		VaultRoot: vaultRoot,
+		HTTP:      &http.Client{Timeout: 30 * time.Second},
+	}
+}
+
+// PairDevice calls POST /api/client/pair and returns device_id and device_token.
+func (c *Client) PairDevice(serverURL, username, password, deviceName, clientVersion string) (deviceID, deviceToken string, err error) {
+	body := map[string]string{
+		"login":          username,
+		"password":       password,
+		"device_name":    deviceName,
+		"client_version": clientVersion,
+	}
+	var resp struct {
+		DeviceID    string `json:"device_id"`
+		DeviceToken string `json:"device_token"`
+	}
+	savedURL := c.ServerURL
+	c.ServerURL = serverURL
+	err = c.post("/api/client/pair", body, &resp)
+	c.ServerURL = savedURL
+	if err != nil {
+		return "", "", err
+	}
+	return resp.DeviceID, resp.DeviceToken, nil
+}
+
+// DeviceInfo holds device information from the server.
+type DeviceInfo struct {
+	DeviceID      string `json:"device_id"`
+	UserID        string `json:"user_id"`
+	Username      string `json:"username"`
+	DeviceName    string `json:"device_name"`
+	ClientVersion string `json:"client_version"`
+	LastSeen      string `json:"last_seen"`
+	RevokedAt     string `json:"revoked_at"`
+	CreatedAt     string `json:"created_at"`
+}
+
+// GetMe calls GET /api/client/me and returns device info.
+func (c *Client) GetMe() (*DeviceInfo, error) {
+	var resp DeviceInfo
+	if err := c.get("/api/client/me", &resp); err != nil {
+		return nil, err
+	}
+	return &resp, nil
+}
+
+// RevokeCurrent calls POST /api/client/revoke-current.
+func (c *Client) RevokeCurrent() error {
+	var resp struct {
+		Status string `json:"status"`
+	}
+	return c.post("/api/client/revoke-current", nil, &resp)
+}
+
+// TestAuth checks credentials without creating a device.
+func (c *Client) TestAuth(serverURL, username, password string) error {
+	body := map[string]string{"username": username, "password": password}
+	savedURL := c.ServerURL
+	savedKey := c.APIKey
+	c.ServerURL = serverURL
+	c.APIKey = ""
+	err := c.post("/api/auth/test", body, nil)
+	c.ServerURL = savedURL
+	c.APIKey = savedKey
+	return err
+}
+
+// PushRequest is the payload for POST /sync/push.
+type PushRequest struct {
+	DeviceID       string   `json:"device_id"`
+	IdempotencyKey string   `json:"idempotency_key,omitempty"`
+	Ops            []PushOp `json:"ops"`
+}
+
+// PushOp is a single operation in a push request.
+type PushOp struct {
+	OpID              string `json:"op_id"`
+	EntityType        string `json:"entity_type"`
+	EntityID          string `json:"entity_id"`
+	OpType            string `json:"op_type"`
+	PayloadJSON       string `json:"payload_json"`
+	ClientSequence    int    `json:"client_sequence"`
+	LastSeenServerSeq int    `json:"last_seen_server_seq"`
+	CreatedAt         string `json:"created_at"`
+}
+
+// PushResponse is the response from POST /sync/push.
+type PushResponse struct {
+	Accepted  []string                 `json:"accepted"`
+	Count     int                      `json:"count"`
+	Conflicts []map[string]interface{} `json:"conflicts"`
+}
+
+// Push sends local operations to the server.
+func (c *Client) Push(ops []Op) (*PushResponse, error) {
+	pushOps := make([]PushOp, len(ops))
+	for i, op := range ops {
+		pushOps[i] = PushOp{
+			OpID:              op.OpID,
+			EntityType:        op.EntityType,
+			EntityID:          op.EntityID,
+			OpType:            op.OpType,
+			PayloadJSON:       op.PayloadJSON,
+			ClientSequence:    op.ClientSequence,
+			LastSeenServerSeq: op.LastSeenServerSeq,
+			CreatedAt:         op.CreatedAt,
+		}
+	}
+	req := PushRequest{DeviceID: c.DeviceID, Ops: pushOps}
+	var resp PushResponse
+	if err := c.post("/api/v1/sync/push", req, &resp); err != nil {
+		return nil, err
+	}
+	return &resp, nil
+}
+
+// PullRequest is the payload for POST /sync/pull.
+type PullRequest struct {
+	SinceSequence int `json:"since_sequence"`
+}
+
+// PullResponse is the response from POST /sync/pull.
+type PullResponse struct {
+	ServerSequence int  `json:"server_sequence"`
+	Ops            []Op `json:"ops"`
+}
+
+// Pull fetches remote operations since a given sequence.
+func (c *Client) Pull(sinceSequence int) (*PullResponse, error) {
+	req := PullRequest{SinceSequence: sinceSequence}
+	var resp PullResponse
+	if err := c.post("/api/v1/sync/pull", req, &resp); err != nil {
+		return nil, err
+	}
+	return &resp, nil
+}
+
+// UploadBlob uploads a file to the server and returns its SHA-256.
+func (c *Client) UploadBlob(localPath string) (sha256 string, err error) {
+	var b bytes.Buffer
+	w := multipart.NewWriter(&b)
+	fw, err := w.CreateFormFile("file", filepath.Base(localPath))
+	if err != nil {
+		return "", err
+	}
+	f, err := os.Open(localPath)
+	if err != nil {
+		return "", err
+	}
+	defer f.Close()
+	if _, err := io.Copy(fw, f); err != nil {
+		return "", err
+	}
+	w.Close()
+
+	req, err := http.NewRequest("POST", c.ServerURL+"/api/v1/blobs/", &b)
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("Content-Type", w.FormDataContentType())
+	req.Header.Set("Authorization", "Bearer "+c.bearerToken())
+
+	resp, err := c.HTTP.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+
+	var result struct {
+		SHA256 string `json:"sha256"`
+		Size   int    `json:"size"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return "", err
+	}
+	return result.SHA256, nil
+}
+
+// DownloadBlob downloads a blob by SHA-256 hash.
+func (c *Client) DownloadBlob(sha256, destPath string) error {
+	req, err := http.NewRequest("GET", c.ServerURL+"/api/v1/blobs/"+sha256, nil)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Authorization", "Bearer "+c.bearerToken())
+
+	resp, err := c.HTTP.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != 200 {
+		return fmt.Errorf("download blob: HTTP %d", resp.StatusCode)
+	}
+
+	out, err := os.Create(destPath)
+	if err != nil {
+		return err
+	}
+	defer out.Close()
+	_, err = io.Copy(out, resp.Body)
+	return err
+}
+
+func (c *Client) bearerToken() string {
+	if c.DeviceToken != "" {
+		return c.DeviceToken
+	}
+	return c.APIKey
+}
+
+func (c *Client) post(path string, body, result interface{}) error {
+	var b bytes.Buffer
+	if body != nil {
+		if err := json.NewEncoder(&b).Encode(body); err != nil {
+			return err
+		}
+	}
+	req, err := http.NewRequest("POST", c.ServerURL+path, &b)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Bearer "+c.bearerToken())
+
+	resp, err := c.HTTP.Do(req)
+	if err != nil {
+		return fmt.Errorf("http: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 400 {
+		data, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("server %d: %s", resp.StatusCode, string(data))
+	}
+
+	if result != nil {
+		return json.NewDecoder(resp.Body).Decode(result)
+	}
+	return nil
+}
+
+func (c *Client) get(path string, result interface{}) error {
+	req, err := http.NewRequest("GET", c.ServerURL+path, nil)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Authorization", "Bearer "+c.bearerToken())
+
+	resp, err := c.HTTP.Do(req)
+	if err != nil {
+		return fmt.Errorf("http: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 400 {
+		data, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("server %d: %s", resp.StatusCode, string(data))
+	}
+
+	if result != nil {
+		return json.NewDecoder(resp.Body).Decode(result)
+	}
+	return nil
+}
@@ -0,0 +1,289 @@
+package sync
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+const (
+	EntityNode    = "node"
+	EntityNote    = "note"
+	EntityFile    = "file"
+	EntityFolder  = "folder"
+	EntityAction  = "action"
+	EntityWorklog = "worklog"
+)
+
+const (
+	OpCreate = "create"
+	OpUpdate = "update"
+	OpDelete = "delete"
+	OpMove   = "move"
+)
+
+// Op represents a sync operation.
+type Op struct {
+	ID                string  `json:"id"`
+	OpID              string  `json:"op_id"`
+	ServerSequence    int     `json:"server_sequence,omitempty"`
+	DeviceID          string  `json:"device_id,omitempty"`
+	EntityType        string  `json:"entity_type"`
+	EntityID          string  `json:"entity_id"`
+	OpType            string  `json:"op_type"`
+	PayloadJSON       string  `json:"payload_json"`
+	CreatedAt         string  `json:"created_at"`
+	PushedAt          *string `json:"pushed_at,omitempty"`
+	AppliedAt         *string `json:"applied_at,omitempty"`
+	ClientSequence    int     `json:"client_sequence,omitempty"`
+	LastSeenServerSeq int     `json:"last_seen_server_seq,omitempty"`
+}
+
+// syncState persists connection state to JSON file.
+type syncState struct {
+	ServerURL   string `json:"server_url"`
+	APIKey      string `json:"api_key"`
+	DeviceID    string `json:"device_id"`
+	LastPullSeq int    `json:"last_pull_seq"`
+	LastSyncAt  string `json:"last_sync_at"`
+}
+
+// Service records and manages sync operations using JSON file storage.
+type Service struct {
+	vaultRoot string
+	deviceID  string
+}
+
+// NewService creates a sync service.
+func NewService(vaultRoot, deviceID string) *Service {
+	return &Service{vaultRoot: vaultRoot, deviceID: deviceID}
+}
+
+func (s *Service) syncDir() string {
+	return filepath.Join(s.vaultRoot, ".verstak", "sync")
+}
+
+func (s *Service) opsPath() string {
+	return filepath.Join(s.syncDir(), "ops.json")
+}
+
+func (s *Service) statePath() string {
+	return filepath.Join(s.syncDir(), "state.json")
+}
+
+func (s *Service) ensureDir() error {
+	return os.MkdirAll(s.syncDir(), 0o755)
+}
+
+// RecordOp writes a sync operation to the local ops file.
+func (s *Service) RecordOp(entityType, entityID, opType string, payload interface{}) error {
+	if err := s.ensureDir(); err != nil {
+		return err
+	}
+	id := uuid.New().String()
+	now := time.Now().UTC().Format(time.RFC3339)
+
+	var payloadStr string
+	if payload != nil {
+		b, err := json.Marshal(payload)
+		if err != nil {
+			return err
+		}
+		payloadStr = string(b)
+	}
+
+	op := Op{
+		ID:         id,
+		OpID:       id,
+		DeviceID:   s.deviceID,
+		EntityType: entityType,
+		EntityID:   entityID,
+		OpType:     opType,
+		PayloadJSON: payloadStr,
+		CreatedAt:  now,
+	}
+
+	ops, err := s.loadOps()
+	if err != nil {
+		return err
+	}
+	ops = append(ops, op)
+	return s.saveOps(ops)
+}
+
+// RecordRemoteOp writes a remote op to the local ops file.
+func (s *Service) RecordRemoteOp(op Op) error {
+	if err := s.ensureDir(); err != nil {
+		return err
+	}
+	now := time.Now().UTC().Format(time.RFC3339)
+
+	ops, err := s.loadOps()
+	if err != nil {
+		return err
+	}
+	remoteID := op.OpID + "-remote"
+	for _, existing := range ops {
+		if existing.ID == remoteID {
+			return nil
+		}
+	}
+	op.ID = remoteID
+	op.PushedAt = &now
+	op.AppliedAt = &now
+	ops = append(ops, op)
+	return s.saveOps(ops)
+}
+
+// GetUnpushedOps returns ops that have not been pushed yet.
+func (s *Service) GetUnpushedOps() ([]Op, error) {
+	ops, err := s.loadOps()
+	if err != nil {
+		return nil, err
+	}
+	var unpushed []Op
+	for _, op := range ops {
+		if op.PushedAt == nil {
+			unpushed = append(unpushed, op)
+		}
+	}
+	return unpushed, nil
+}
+
+// MarkPushed marks ops as pushed to server.
+func (s *Service) MarkPushed(opIDs []string) error {
+	ops, err := s.loadOps()
+	if err != nil {
+		return err
+	}
+	now := time.Now().UTC().Format(time.RFC3339)
+	idSet := make(map[string]bool, len(opIDs))
+	for _, id := range opIDs {
+		idSet[id] = true
+	}
+	for i := range ops {
+		if idSet[ops[i].OpID] {
+			ops[i].PushedAt = &now
+		}
+	}
+	return s.saveOps(ops)
+}
+
+// MarkApplied marks remote ops as applied locally.
+func (s *Service) MarkApplied(opIDs []string) error {
+	ops, err := s.loadOps()
+	if err != nil {
+		return err
+	}
+	now := time.Now().UTC().Format(time.RFC3339)
+	idSet := make(map[string]bool, len(opIDs))
+	for _, id := range opIDs {
+		idSet[id] = true
+	}
+	for i := range ops {
+		if idSet[ops[i].OpID] {
+			ops[i].AppliedAt = &now
+		}
+	}
+	return s.saveOps(ops)
+}
+
+// GetState returns the current sync state.
+func (s *Service) GetState() (serverURL, apiKey string, lastPullSeq int, lastSyncAt string, err error) {
+	st, err := s.loadState()
+	if err != nil {
+		return "", "", 0, "", err
+	}
+	return st.ServerURL, st.APIKey, st.LastPullSeq, st.LastSyncAt, nil
+}
+
+// SetState saves sync connection state.
+func (s *Service) SetState(serverURL, apiKey string) error {
+	if err := s.ensureDir(); err != nil {
+		return err
+	}
+	st, err := s.loadState()
+	if err != nil {
+		st = &syncState{}
+	}
+	st.ServerURL = serverURL
+	st.APIKey = apiKey
+	return s.saveState(st)
+}
+
+// SetLastPullSeq updates the last pulled server sequence.
+func (s *Service) SetLastPullSeq(seq int) error {
+	st, err := s.loadState()
+	if err != nil {
+		return err
+	}
+	st.LastPullSeq = seq
+	return s.saveState(st)
+}
+
+// SetLastSyncAt updates the last sync timestamp.
+func (s *Service) SetLastSyncAt(t string) error {
+	st, err := s.loadState()
+	if err != nil {
+		return err
+	}
+	st.LastSyncAt = t
+	return s.saveState(st)
+}
+
+// GetDeviceID returns the device ID used by this service.
+func (s *Service) GetDeviceID() string {
+	return s.deviceID
+}
+
+// --- file helpers ---
+
+func (s *Service) loadOps() ([]Op, error) {
+	data, err := os.ReadFile(s.opsPath())
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("read ops: %w", err)
+	}
+	var ops []Op
+	if err := json.Unmarshal(data, &ops); err != nil {
+		return nil, fmt.Errorf("parse ops: %w", err)
+	}
+	return ops, nil
+}
+
+func (s *Service) saveOps(ops []Op) error {
+	data, err := json.MarshalIndent(ops, "", "  ")
+	if err != nil {
+		return fmt.Errorf("marshal ops: %w", err)
+	}
+	return os.WriteFile(s.opsPath(), data, 0o644)
+}
+
+func (s *Service) loadState() (*syncState, error) {
+	data, err := os.ReadFile(s.statePath())
+	if err != nil {
+		if os.IsNotExist(err) {
+			return &syncState{}, nil
+		}
+		return nil, fmt.Errorf("read state: %w", err)
+	}
+	var st syncState
+	if err := json.Unmarshal(data, &st); err != nil {
+		return nil, fmt.Errorf("parse state: %w", err)
+	}
+	return &st, nil
+}
+
+func (s *Service) saveState(st *syncState) error {
+	data, err := json.MarshalIndent(st, "", "  ")
+	if err != nil {
+		return fmt.Errorf("marshal state: %w", err)
+	}
+	return os.WriteFile(s.statePath(), data, 0o644)
+}