feat: sync — client ops recording in core services
- internal/core/sync/: Service, Client, Blob packages - RecordOp creates sync_ops entries for all mutations - Client for push/pull/blob HTTP to server - Blob SHA-256 hashing and local storage - Wired into app.go alongside activity recording - Device ID from config or fallback
This commit is contained in:
@@ -0,0 +1,80 @@
|
||||
package sync
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
// BlobDir returns the path to .verstak/blobs/ inside the vault.
|
||||
func BlobDir(vaultRoot string) string {
|
||||
return filepath.Join(vaultRoot, ".verstak", "blobs")
|
||||
}
|
||||
|
||||
// BlobPath returns the on-disk path for a SHA-256 hash.
|
||||
func BlobPath(blobsDir, shaHex string) string {
|
||||
return filepath.Join(blobsDir, shaHex[:2], shaHex[2:4], shaHex)
|
||||
}
|
||||
|
||||
// HashFile computes SHA-256 of a file.
|
||||
func HashFile(path string) (string, error) {
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer f.Close()
|
||||
h := sha256.New()
|
||||
if _, err := io.Copy(h, f); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return hex.EncodeToString(h.Sum(nil)), nil
|
||||
}
|
||||
|
||||
// HashBytes computes SHA-256 of byte data.
|
||||
func HashBytes(data []byte) string {
|
||||
h := sha256.Sum256(data)
|
||||
return hex.EncodeToString(h[:])
|
||||
}
|
||||
|
||||
// StoreBlob copies a file into the blob store, returns its SHA-256.
|
||||
func StoreBlob(blobsDir, srcPath string) (string, error) {
|
||||
shaHex, err := HashFile(srcPath)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
dest := BlobPath(blobsDir, shaHex)
|
||||
if _, err := os.Stat(dest); err == nil {
|
||||
return shaHex, nil // already exists
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(dest), 0750); err != nil {
|
||||
return "", err
|
||||
}
|
||||
src, err := os.Open(srcPath)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer src.Close()
|
||||
|
||||
dst, err := os.Create(dest)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer dst.Close()
|
||||
|
||||
if _, err := io.Copy(dst, src); err != nil {
|
||||
return "", err
|
||||
}
|
||||
_ = dst.Sync()
|
||||
return shaHex, nil
|
||||
}
|
||||
|
||||
// ReadBlob reads a blob by SHA-256 hash.
|
||||
func ReadBlob(blobsDir, shaHex string) ([]byte, error) {
|
||||
return os.ReadFile(BlobPath(blobsDir, shaHex))
|
||||
}
|
||||
|
||||
// Ensure the package compiles without unused errors.
|
||||
var _ = fmt.Sprintf
|
||||
@@ -0,0 +1,209 @@
|
||||
package sync
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"mime/multipart"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Client communicates with the Verstak Sync Server.
|
||||
type Client struct {
|
||||
ServerURL string
|
||||
APIKey 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},
|
||||
}
|
||||
}
|
||||
|
||||
// RegisterDevice calls POST /api/v1/device/register and returns the API key.
|
||||
func (c *Client) RegisterDevice(name string) (apiKey string, err error) {
|
||||
body := map[string]string{"name": name}
|
||||
var resp struct {
|
||||
DeviceID string `json:"device_id"`
|
||||
APIKey string `json:"api_key"`
|
||||
}
|
||||
if err := c.post("/api/v1/device/register", body, &resp); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return resp.APIKey, nil
|
||||
}
|
||||
|
||||
// PushRequest is the payload for POST /sync/push.
|
||||
type PushRequest struct {
|
||||
DeviceID string `json:"device_id"`
|
||||
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"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
}
|
||||
|
||||
// PushResponse is the response from POST /sync/push.
|
||||
type PushResponse struct {
|
||||
Accepted []string `json:"accepted"`
|
||||
Count int `json:"count"`
|
||||
}
|
||||
|
||||
// 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,
|
||||
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 {
|
||||
SinceRevision int `json:"since_revision"`
|
||||
}
|
||||
|
||||
// PullResponse is the response from POST /sync/pull.
|
||||
type PullResponse struct {
|
||||
ServerRevision int `json:"server_revision"`
|
||||
Ops []Op `json:"ops"`
|
||||
}
|
||||
|
||||
// Pull fetches remote operations since a given revision.
|
||||
func (c *Client) Pull(sinceRevision int) (*PullResponse, error) {
|
||||
req := PullRequest{SinceRevision: sinceRevision}
|
||||
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.APIKey)
|
||||
|
||||
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.APIKey)
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
// --- internal ---
|
||||
|
||||
func (c *Client) post(path string, body, result interface{}) error {
|
||||
var b bytes.Buffer
|
||||
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.APIKey)
|
||||
|
||||
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,155 @@
|
||||
package sync
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"verstak/internal/core/storage"
|
||||
"verstak/internal/core/util"
|
||||
)
|
||||
|
||||
// Entity types (matches activity targets).
|
||||
const (
|
||||
EntityNode = "node"
|
||||
EntityNote = "note"
|
||||
EntityFile = "file"
|
||||
EntityFolder = "folder"
|
||||
EntityAction = "action"
|
||||
EntityWorklog = "worklog"
|
||||
)
|
||||
|
||||
// Op types.
|
||||
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"`
|
||||
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"`
|
||||
}
|
||||
|
||||
// Service records and manages sync operations.
|
||||
type Service struct {
|
||||
db *storage.DB
|
||||
deviceID string
|
||||
}
|
||||
|
||||
// NewService creates a sync service.
|
||||
func NewService(db *storage.DB, deviceID string) *Service {
|
||||
return &Service{db: db, deviceID: deviceID}
|
||||
}
|
||||
|
||||
// RecordOp writes a sync operation to the local sync_ops table.
|
||||
func (s *Service) RecordOp(entityType, entityID, opType string, payload interface{}) error {
|
||||
id := util.UUID7()
|
||||
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)
|
||||
}
|
||||
|
||||
_, err := s.db.Exec(
|
||||
`INSERT INTO sync_ops (id, op_id, device_id, entity_type, entity_id, op_type, payload_json, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
id, id, s.deviceID, entityType, entityID, opType, payloadStr, now,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// GetUnpushedOps returns ops that have not been pushed yet.
|
||||
func (s *Service) GetUnpushedOps() ([]Op, error) {
|
||||
rows, err := s.db.Query(
|
||||
`SELECT id, op_id, device_id, entity_type, entity_id, op_type, payload_json, created_at, pushed_at
|
||||
FROM sync_ops WHERE pushed_at IS NULL ORDER BY created_at`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
return scanOps(rows)
|
||||
}
|
||||
|
||||
// MarkPushed marks ops as pushed to server.
|
||||
func (s *Service) MarkPushed(opIDs []string) error {
|
||||
now := time.Now().UTC().Format(time.RFC3339)
|
||||
for _, id := range opIDs {
|
||||
_, err := s.db.Exec("UPDATE sync_ops SET pushed_at=? WHERE op_id=?", now, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// MarkApplied marks remote ops as applied locally.
|
||||
func (s *Service) MarkApplied(opIDs []string) error {
|
||||
now := time.Now().UTC().Format(time.RFC3339)
|
||||
for _, id := range opIDs {
|
||||
_, err := s.db.Exec("UPDATE sync_ops SET applied_at=? WHERE op_id=?", now, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetState returns the current sync state.
|
||||
func (s *Service) GetState() (serverURL, apiKey string, lastPushRev int, lastSyncAt string, err error) {
|
||||
err = s.db.QueryRow(
|
||||
`SELECT server_url, api_key, last_push_rev, COALESCE(last_sync_at,'') FROM sync_state WHERE device_id=?`,
|
||||
s.deviceID).Scan(&serverURL, &apiKey, &lastPushRev, &lastSyncAt)
|
||||
if err == sql.ErrNoRows {
|
||||
return "", "", 0, "", nil
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// SetState saves sync connection state.
|
||||
func (s *Service) SetState(serverURL, apiKey string) error {
|
||||
_, err := s.db.Exec(
|
||||
`INSERT INTO sync_state (device_id, server_url, api_key, last_push_rev, last_sync_at)
|
||||
VALUES (?, ?, ?, 0, '')
|
||||
ON CONFLICT(device_id) DO UPDATE SET server_url=excluded.server_url, api_key=excluded.api_key`,
|
||||
s.deviceID, serverURL, apiKey,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// --- helpers ---
|
||||
|
||||
func scanOps(rows *sql.Rows) ([]Op, error) {
|
||||
var out []Op
|
||||
for rows.Next() {
|
||||
var o Op
|
||||
var pushedAt sql.NullString
|
||||
if err := rows.Scan(&o.ID, &o.OpID, &o.DeviceID, &o.EntityType, &o.EntityID,
|
||||
&o.OpType, &o.PayloadJSON, &o.CreatedAt, &pushedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if pushedAt.Valid {
|
||||
o.PushedAt = &pushedAt.String
|
||||
}
|
||||
out = append(out, o)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// MustVar ensures the package is not considered unused.
|
||||
var _ = fmt.Sprintf
|
||||
Reference in New Issue
Block a user