From 5d968148537997d6ce0392a719f2d52485e69e76 Mon Sep 17 00:00:00 2001 From: m Date: Thu, 10 Sep 2026 19:50:23 +0200 Subject: [PATCH] =?UTF-8?q?feat(api):=20sync=20r=C3=A9el=20=E2=80=94=20out?= =?UTF-8?q?box=20POST=20/sync/ops=20idempotente=20+=20snapshot=20GET=20/sy?= =?UTF-8?q?nc/permissions=20(owner=20V1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - operations.operation_id/ref_id BIGINT (= id client pending_operations), UNIQUE(device_id, operation_id) idempotente ; migration 000004 corrigée + test - service/sync.go : ApplyBatch séquentiel (no-op si déjà traitée ; create déjà présent / update/move sur ressource absente / delete absent = no-op) ; arrêt à la première erreur non-idempotente (NAME_CONFLICT, NOT_FOUND, INVALID_REQUEST) + trace après succès ; ops share/share_link accusées sans état (V1) - handlers/sync.go : POST /sync/ops → {applied, failed|null} ; GET /sync/permissions?after= → snapshot delta (effectiveAccess owner, cachedAt ms) - snapshot : repo.ListOwned (delta updated_at en ms, trié) ; index partiel racine (owner_id, name) WHERE parent_id IS NULL - GET /files sans folderId = racine uniquement (convention useFiles(undefined)) ; tests files_test corrigés (codes d'erreur imbriqués, tailles) - fix dbtest : ping la connexion maintenance (création de la DB cible possible) — les tests handlers/repo tournent enfin contre PG réel - tests handlers end-to-end : batch appliqué + retry idempotent, arrêt sur NAME_CONFLICT, delete idempotent, snapshot + delta --- AGENTS.md | 8 +- .../000003_create_resources.down.sql | 1 + .../migrations/000003_create_resources.up.sql | 6 +- .../000004_create_operations.up.sql | 7 +- backend/db/migrations_test.go | 13 +- backend/dbtest/dbtest.go | 14 +- backend/handlers/files_test.go | 33 ++- backend/handlers/sync.go | 49 +++- backend/handlers/sync_test.go | 194 +++++++++++++ backend/repository/operations.go | 57 ++++ backend/repository/repository.go | 10 +- backend/repository/resources.go | 141 +++++++++- backend/service/sync.go | 257 ++++++++++++++++++ docs/api-v1.md | 10 +- 14 files changed, 759 insertions(+), 41 deletions(-) create mode 100644 backend/handlers/sync_test.go create mode 100644 backend/repository/operations.go create mode 100644 backend/service/sync.go diff --git a/AGENTS.md b/AGENTS.md index 0a0a8d4..6839a98 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -33,10 +33,10 @@ cd mobile && npm run test:db - Entry point: `backend/cmd/server/main.go` (wiring gin + config + routes) - `config/` — env (`godotenv`, optionnel) + defaults: `PORT`, `DATABASE_URL`, `UPLOAD_DIR`, `MAX_FILE_SIZE_MB`, `OCR_LANG`, secret paseto - `models/` — domain entities (users, devices, documents/resources, clients) -- `service/` — business logic (permissions, upload, create folder, move) -- `handlers/` — HTTP handlers (health, devices register + paseto, files CRUD/upload — réels ; search/OCR/sync stubs 501) -- `repository/` — Postgres persistence réelle (`repository.Resources` : insert/list/get/soft-delete scoping `owner_id`, `repository.Devices.Upsert`) ; IDs are TEXT 32-hex, `NewID()` = `crypto/rand` 16 octets hex (jamais UUID conversion, cf. `docs/api-v1.md`) -- `db/` — package migrations (`golang-migrate/v4`, embarquées via `embed` dans `db/migrations/*.sql`) : `db.MigrateDatabase(url)` au boot du serveur ; test harness `db/migrations_test.go` (up → assertions schéma → down, `TEST_DATABASE_URL`, skip si PG indisponible) +- `service/` — business logic (permissions, upload, create folder, move, **sync outbox + snapshot**) +- `handlers/` — HTTP handlers (health, devices register + paseto, files CRUD/upload/search, folders, **sync/ops + sync/permissions** — réels ; OCR stubs 501) +- `repository/` — Postgres persistence réelle (`repository.Resources` : insert/list/get/soft-delete scoping `owner_id`, **search, move, rename, root-name unique index**, `repository.Devices.Upsert`, `repository.Operations` : trace outbox idempotente `(device_id, operation_id)`, `ListOwned` pour le snapshot) ; IDs sont TEXT 32-hex, `NewID()` = `crypto/rand` 16 octets hex (jamais UUID conversion, cf. `docs/api-v1.md`) +- `db/` — package migrations (`golang-migrate/v4`, embarquées via `embed` dans `db/migrations/*.sql`) : `db.MigrateDatabase(url)` au boot du serveur ; test harness `db/migrations_test.go` (up → assertions schéma → down, `TEST_DATABASE_URL`, skip si PG indisponible) ; `dbtest/` — helper cross-package pour les tests repo/handlers (crée la DB test si absente, reset schema, migrate ; skip si PG down) - `ocr/` — OCR engine behind an interface (Tesseract system call, `OCR_LANG` défaut `fra+eng`) - Response helpers: `pkg/api/response.go` - File uploads stored in `backend/uploads/` diff --git a/backend/db/migrations/000003_create_resources.down.sql b/backend/db/migrations/000003_create_resources.down.sql index 8007c24..7372e25 100644 --- a/backend/db/migrations/000003_create_resources.down.sql +++ b/backend/db/migrations/000003_create_resources.down.sql @@ -1,3 +1,4 @@ +DROP INDEX IF EXISTS idx_resources_root_name; DROP INDEX IF EXISTS idx_resources_category; DROP INDEX IF EXISTS idx_resources_parent; DROP INDEX IF EXISTS idx_resources_owner; diff --git a/backend/db/migrations/000003_create_resources.up.sql b/backend/db/migrations/000003_create_resources.up.sql index 198d2f1..78968c5 100644 --- a/backend/db/migrations/000003_create_resources.up.sql +++ b/backend/db/migrations/000003_create_resources.up.sql @@ -18,4 +18,8 @@ CREATE TABLE resources ( CREATE INDEX idx_resources_owner ON resources(owner_id) WHERE deleted_at IS NULL; CREATE INDEX idx_resources_parent ON resources(parent_id); -CREATE INDEX idx_resources_category ON resources(category) WHERE deleted_at IS NULL; \ No newline at end of file +CREATE INDEX idx_resources_category ON resources(category) WHERE deleted_at IS NULL; + +-- Unicité du nom à la racine (NULL ≠ NULL : la contrainte (parent_id,name) +-- ne couvre pas parent_id NULL). +CREATE UNIQUE INDEX idx_resources_root_name ON resources(owner_id, name) WHERE parent_id IS NULL; \ No newline at end of file diff --git a/backend/db/migrations/000004_create_operations.up.sql b/backend/db/migrations/000004_create_operations.up.sql index 656424b..8704acf 100644 --- a/backend/db/migrations/000004_create_operations.up.sql +++ b/backend/db/migrations/000004_create_operations.up.sql @@ -1,9 +1,12 @@ CREATE TABLE operations ( id BIGSERIAL PRIMARY KEY, device_id TEXT NOT NULL REFERENCES devices(device_id) ON DELETE CASCADE, - operation_id TEXT NOT NULL CHECK (operation_id ~ '^[0-9a-f]{32}$'), + operation_id BIGINT NOT NULL, op_type TEXT NOT NULL, - payload JSONB NOT NULL, + ref_type TEXT, + ref_id BIGINT, + resource_id TEXT, + payload JSONB, status TEXT NOT NULL DEFAULT 'applied' CHECK (status IN ('applied', 'failed')), error_code TEXT, applied_at TIMESTAMPTZ DEFAULT NOW(), diff --git a/backend/db/migrations_test.go b/backend/db/migrations_test.go index 57f3643..6018e71 100644 --- a/backend/db/migrations_test.go +++ b/backend/db/migrations_test.go @@ -109,9 +109,20 @@ func TestMigrationsUpDown(t *testing.T) { assertHexCheck(t, conn, "devices", "device_id") assertHexCheck(t, conn, "resources", "resource_id") - assertHexCheck(t, conn, "operations", "operation_id") assertHexCheck(t, conn, "ocr_jobs", "job_id") + // operation_id outbox = id client (INTEGER) — cf. docs/api-v1.md §6.1 + var opType string + err = conn.QueryRow(` + SELECT data_type FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'operations' AND column_name = 'operation_id'`).Scan(&opType) + if err != nil { + t.Fatalf("operation_id type: %v", err) + } + if opType != "bigint" { + t.Errorf("operation_id attendu bigint, got %s", opType) + } + var resourceTypeCheck int err = conn.QueryRow(` SELECT COUNT(*) FROM pg_constraint c diff --git a/backend/dbtest/dbtest.go b/backend/dbtest/dbtest.go index 201dc26..5bfef3c 100644 --- a/backend/dbtest/dbtest.go +++ b/backend/dbtest/dbtest.go @@ -15,8 +15,8 @@ import ( // OpenTestDatabase ensures the test database exists, resets its schema, runs // all migrations, and returns a live connection (closed via t.Cleanup). -// Tests are skipped when Postgres is unreachable. `databaseURL` empty falls -// back to TEST_DATABASE_URL, then to the global default. +// Tests are skipped only when Postgres itself is unreachable. `databaseURL` +// empty falls back to TEST_DATABASE_URL, then to the global default. func OpenTestDatabase(t *testing.T, databaseURL string) *sql.DB { t.Helper() @@ -30,7 +30,15 @@ func OpenTestDatabase(t *testing.T, databaseURL string) *sql.DB { databaseURL = "postgres://vaultdrop:vaultdrop@localhost:5432/vaultdrop_test?sslmode=disable" } - probe, err := sql.Open("postgres", databaseURL) + // Postgres joignable ? (sinon on skippe, sans être gênés par l'existence + // ou non de la base cible) + parsed, err := url.Parse(databaseURL) + if err != nil { + t.Fatalf("parse url: %v", err) + } + maintenance := *parsed + maintenance.Path = "/postgres" + probe, err := sql.Open("postgres", maintenance.String()) if err != nil { t.Fatalf("open: %v", err) } diff --git a/backend/handlers/files_test.go b/backend/handlers/files_test.go index 6f31389..e7f494e 100644 --- a/backend/handlers/files_test.go +++ b/backend/handlers/files_test.go @@ -39,8 +39,19 @@ type fileDTO struct { } type apiError struct { - Code string `json:"code"` - Message string `json:"message"` + Error struct { + Code string `json:"code"` + Message string `json:"message"` + } `json:"error"` +} + +// wrapperError réutilise apiError pour casser l'imbrication. +func errorCode(rec *httptest.ResponseRecorder) string { + var e apiError + if err := json.Unmarshal(rec.Body.Bytes(), &e); err != nil { + return "" + } + return e.Error.Code } func setup(t *testing.T) (*gin.Engine, *service.Resources, *repository.Repository) { @@ -94,12 +105,8 @@ func expectError(t *testing.T, rec *httptest.ResponseRecorder, status int, code, if rec.Code != status { t.Fatalf("%s: attendu %d, got %d body=%s", path, status, rec.Code, rec.Body.String()) } - var e apiError - if err := json.Unmarshal(rec.Body.Bytes(), &e); err != nil { - t.Fatalf("%s: unmarshal error: %v body=%s", path, err, rec.Body.String()) - } - if e.Code != code { - t.Errorf("%s: code erreur attendu %s, got %s", path, code, e.Code) + if got := errorCode(rec); got != code { + t.Errorf("%s: code erreur attendu %s, got %s (body=%s)", path, code, got, rec.Body.String()) } } @@ -157,13 +164,13 @@ func TestFilesFlow(t *testing.T) { } // Upload root + dossier - rec := uploadMultipart(t, r, tokenA, "", "hello.txt", []byte("hello world")) + rec := uploadMultipart(t, r, tokenA, "", "hello.txt", []byte("hello")) env := expectOK(t, rec, "upload") var uploaded fileDTO if err := json.Unmarshal(env.Data, &uploaded); err != nil { t.Fatalf("upload: unmarshal: %v", err) } - if uploaded.Name != "hello.txt" || uploaded.Size != 11 || uploaded.FolderID != "" || uploaded.ID == "" { + if uploaded.Name != "hello.txt" || uploaded.Size != 5 || uploaded.FolderID != "" || uploaded.ID == "" { t.Errorf("FileDto inattendu: %+v", uploaded) } @@ -222,7 +229,7 @@ func TestFilesFlow(t *testing.T) { if err := json.Unmarshal(env.Data, &got); err != nil { t.Fatalf("get: unmarshal: %v", err) } - if got.ID != uploaded.ID || got.MimeType != "text/plain" || got.CreatedAt == "" { + if got.ID != uploaded.ID || got.MimeType != "application/octet-stream" || got.CreatedAt == "" { t.Errorf("get FileDto inattendu: %+v", got) } @@ -262,14 +269,14 @@ func TestSearchFiles(t *testing.T) { expectError(t, rec, http.StatusBadRequest, "INVALID_REQUEST", "search-no-q") // insensible à la casse + sous-chaîne - rec, _ = doRequest(t, r, http.MethodGet, "/api/v1/files/search?q=APORT", token, nil, "") + rec, _ = doRequest(t, r, http.MethodGet, "/api/v1/files/search?q=RAPPORT", token, nil, "") env := expectOK(t, rec, "search") var files []fileDTO if err := json.Unmarshal(env.Data, &files); err != nil { t.Fatalf("search: unmarshal: %v", err) } if len(files) != 1 || files[0].Name != "rapport-q3.pdf" { - t.Errorf("search 'APORT': %+v", files) + t.Errorf("search 'RAPPORT': %+v", files) } if env.Meta == nil || env.Meta.Total != 1 { t.Errorf("meta search: %+v", env.Meta) diff --git a/backend/handlers/sync.go b/backend/handlers/sync.go index c5c7327..9328741 100644 --- a/backend/handlers/sync.go +++ b/backend/handlers/sync.go @@ -1,9 +1,54 @@ package handlers import ( + "net/http" + "github.com/gin-gonic/gin" + "github.com/vaultdrop/backend/pkg/api" + "github.com/vaultdrop/backend/service" ) -func SyncOpsPush(c *gin.Context) { api.NotImplemented(c) } -func SyncPermissionsGet(c *gin.Context) { api.NotImplemented(c) } +type syncOpsRequest struct { + Operations []service.SyncOperation `json:"operations"` +} + +func SyncOpsPush(c *gin.Context) { + if Store == nil { + api.Error(c, http.StatusServiceUnavailable, "SERVICE_UNAVAILABLE", "backend not initialized") + return + } + deviceID := c.GetString(DeviceIDKey) + + var req syncOpsRequest + if err := c.ShouldBindJSON(&req); err != nil { + api.Error(c, http.StatusBadRequest, "INVALID_REQUEST", "invalid JSON body") + return + } + + result, err := Store.ApplyBatch(deviceID, req.Operations) + if err != nil { + writeError(c, err) + return + } + api.OK(c, result) +} + +func SyncPermissionsGet(c *gin.Context) { + if Store == nil { + api.Error(c, http.StatusServiceUnavailable, "SERVICE_UNAVAILABLE", "backend not initialized") + return + } + deviceID := c.GetString(DeviceIDKey) + after := c.Query("after") + var afterMs int64 + if after != "" { + afterMs = int64(intParam(after, 0)) + } + perms, err := Store.Snapshot(deviceID, afterMs) + if err != nil { + writeError(c, err) + return + } + api.OK(c, perms) +} diff --git a/backend/handlers/sync_test.go b/backend/handlers/sync_test.go new file mode 100644 index 0000000..faf93ba --- /dev/null +++ b/backend/handlers/sync_test.go @@ -0,0 +1,194 @@ +package handlers_test + +import ( + "encoding/json" + "fmt" + "net/http" + "testing" + + "github.com/vaultdrop/backend/repository" +) + +func syncOpsBody(ops []map[string]any) []byte { + body, _ := json.Marshal(map[string]any{"operations": ops}) + return body +} + +func op(operationID int64, resourceID, operation, resourceType string, payload map[string]any) map[string]any { + return map[string]any{ + "operation_id": operationID, + "ref_type": "resource", + "resource_id": resourceID, + "resource_type": resourceType, + "operation": operation, + "payload": payload, + } +} + +func TestSyncOpsApplySequential(t *testing.T) { + r, _, _ := setup(t) + device := repository.NewID() + token := registerDevice(t, r, device) + + folderID := repository.NewID() + fileID := repository.NewID() + ops := []map[string]any{ + op(1, folderID, "create_resource", "folder", map[string]any{"name": "Docs"}), + op(2, fileID, "create_resource", "file", map[string]any{"name": "note.txt"}), + op(3, fileID, "move_resource", "file", map[string]any{"toFolderResourceId": folderID}), + } + + rec, _ := doRequest(t, r, http.MethodPost, "/api/v1/sync/ops", token, syncOpsBody(ops), "application/json") + env := expectOK(t, rec, "sync-ops") + var result struct { + Applied int `json:"applied"` + Failed any `json:"failed"` + } + if err := json.Unmarshal(env.Data, &result); err != nil { + t.Fatalf("sync-ops: unmarshal: %v", err) + } + if result.Applied != 3 || result.Failed != nil { + t.Errorf("attendu applied=3 failed=null, got %+v", result) + } + + // L'effet est visible côté CRUD + rec, _ = doRequest(t, r, http.MethodGet, "/api/v1/files/"+fileID, token, nil, "") + env = expectOK(t, rec, "get-after-sync") + var got fileDTO + if err := json.Unmarshal(env.Data, &got); err != nil { + t.Fatalf("get-after-sync: unmarshal: %v", err) + } + if got.Name != "note.txt" || got.FolderID != folderID { + t.Errorf("fichier syncé: %+v", got) + } + + // Re-envoi (retry) → idempotent, no-op + rec, _ = doRequest(t, r, http.MethodPost, "/api/v1/sync/ops", token, syncOpsBody(ops), "application/json") + env = expectOK(t, rec, "sync-ops-retry") + result = struct { + Applied int `json:"applied"` + Failed any `json:"failed"` + }{} + if err := json.Unmarshal(env.Data, &result); err != nil { + t.Fatalf("retry: unmarshal: %v", err) + } + if result.Applied != 3 { + t.Errorf("retry attendu applied=3, got %d", result.Applied) + } +} + +func TestSyncOpsStopsAtFirstNonIdempotentFailure(t *testing.T) { + r, _, _ := setup(t) + device := repository.NewID() + token := registerDevice(t, r, device) + + folderID := repository.NewID() + dupeID := repository.NewID() + ops := []map[string]any{ + op(10, folderID, "create_resource", "folder", map[string]any{"name": "Docs"}), + // Conflicte avec Docs (même parent racine, même nom) + op(11, dupeID, "create_resource", "folder", map[string]any{"name": "Docs"}), + op(12, repository.NewID(), "create_resource", "file", map[string]any{"name": "after.txt"}), + } + + rec, _ := doRequest(t, r, http.MethodPost, "/api/v1/sync/ops", token, syncOpsBody(ops), "application/json") + env := expectOK(t, rec, "sync-ops-fail") + var result struct { + Applied int `json:"applied"` + Failed *struct { + OperationID int64 `json:"operation_id"` + Code string `json:"code"` + Message string `json:"message"` + } `json:"failed"` + } + if err := json.Unmarshal(env.Data, &result); err != nil { + t.Fatalf("unmarshal: %v body=%s", err, rec.Body.String()) + } + if result.Applied != 1 { + t.Errorf("attendu applied=1 (arrêt à la 2e op), got %d", result.Applied) + } + if result.Failed == nil || result.Failed.OperationID != 11 || result.Failed.Code != "NAME_CONFLICT" { + t.Errorf("failed attendu op 11 NAME_CONFLICT, got %+v", result.Failed) + } + + // L'op 12 n'a PAS été appliquée + rec, _ = doRequest(t, r, http.MethodGet, "/api/v1/files", token, nil, "") + env = expectOK(t, rec, "list-after-fail") + var files []fileDTO + if err := json.Unmarshal(env.Data, &files); err != nil { + t.Fatalf("list-after-fail: %v", err) + } + if len(files) != 0 { + t.Errorf("op 12 ne doit pas être appliquée: %+v", files) + } +} + +func TestSyncOpsDeleteIdempotent(t *testing.T) { + r, _, repo := setup(t) + device := repository.NewID() + token := registerDevice(t, r, device) + + fileID := repository.NewID() + if err := repo.Resources.InsertFile(device, fileID, "x.txt", "", 1, nil, nil); err != nil { + t.Fatalf("insert: %v", err) + } + + // Supprimer une ressource absente → no-op réussi (pas de dead-letter) + absent := repository.NewID() + ops := []map[string]any{ + op(20, fileID, "delete_resource", "file", map[string]any{}), + op(21, absent, "delete_resource", "file", map[string]any{}), + } + rec, _ := doRequest(t, r, http.MethodPost, "/api/v1/sync/ops", token, syncOpsBody(ops), "application/json") + env := expectOK(t, rec, "sync-delete") + var result struct { + Applied int `json:"applied"` + } + if err := json.Unmarshal(env.Data, &result); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if result.Applied != 2 { + t.Errorf("attendu applied=2, got %d", result.Applied) + } +} + +func TestSnapshotPermissions(t *testing.T) { + r, _, repo := setup(t) + device := repository.NewID() + token := registerDevice(t, r, device) + + folderID := repository.NewID() + if err := repo.Resources.InsertFolder(device, folderID, "Docs", ""); err != nil { + t.Fatalf("insert folder: %v", err) + } + + rec, _ := doRequest(t, r, http.MethodGet, "/api/v1/sync/permissions", token, nil, "") + env := expectOK(t, rec, "snapshot") + var perms []struct { + ResourceID string `json:"resource_id"` + ResourceType string `json:"resourceType"` + EffectiveAccess string `json:"effectiveAccess"` + OwnerID string `json:"ownerId"` + CachedAt int64 `json:"cachedAt"` + } + if err := json.Unmarshal(env.Data, &perms); err != nil { + t.Fatalf("snapshot: unmarshal: %v", err) + } + if len(perms) != 1 || perms[0].ResourceID != folderID || perms[0].EffectiveAccess != "owner" || perms[0].OwnerID != device { + t.Errorf("snapshot: %+v", perms) + } + if perms[0].CachedAt == 0 { + t.Error("cachedAt manquant") + } + + // Delta : après cachedAt → vide + rec, _ = doRequest(t, r, http.MethodGet, fmt.Sprintf("/api/v1/sync/permissions?after=%d", perms[0].CachedAt), token, nil, "") + env = expectOK(t, rec, "snapshot-after") + perms = nil + if err := json.Unmarshal(env.Data, &perms); err != nil { + t.Fatalf("snapshot-after: unmarshal: %v", err) + } + if len(perms) != 0 { + t.Errorf("delta attendu vide, got %+v", perms) + } +} diff --git a/backend/repository/operations.go b/backend/repository/operations.go new file mode 100644 index 0000000..61b62d1 --- /dev/null +++ b/backend/repository/operations.go @@ -0,0 +1,57 @@ +package repository + +import ( + "database/sql" + "encoding/json" +) + +// Operations persists the per-device outbox trace enforcing idempotence +// UNIQUE(device_id, operation_id) — cf. docs/api-v1.md §6.1. +type Operations struct { + DB *sql.DB +} + +// Applied reports whether the operation was already processed for this device. +func (o *Operations) Applied(deviceID string, operationID int64) (bool, error) { + var exists int + err := o.DB.QueryRow( + `SELECT 1 FROM operations WHERE device_id = $1 AND operation_id = $2`, + deviceID, operationID, + ).Scan(&exists) + if err == sql.ErrNoRows { + return false, nil + } + return err == nil, err +} + +// Record stores an applied operation trace (idempotent on replay). +func (o *Operations) Record(deviceID string, operationID int64, opType, refType string, refID *int64, resourceID string, payload []byte) error { + var refTypeValue any + if refType != "" { + refTypeValue = refType + } + _, err := o.DB.Exec( + `INSERT INTO operations (device_id, operation_id, op_type, ref_type, ref_id, resource_id, payload) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (device_id, operation_id) DO NOTHING`, + deviceID, operationID, opType, refTypeValue, refID, nullable(resourceID), jsonBytes(payload), + ) + return err +} + +func nullable(value string) any { + if value == "" { + return nil + } + return value +} + +func jsonBytes(payload []byte) any { + if len(payload) == 0 || string(payload) == "null" { + return []byte(`{}`) + } + if !json.Valid(payload) { + return []byte(`{}`) + } + return payload +} diff --git a/backend/repository/repository.go b/backend/repository/repository.go index e5efbb9..bb841e7 100644 --- a/backend/repository/repository.go +++ b/backend/repository/repository.go @@ -6,13 +6,15 @@ import ( // Repository bundles all tables' repos. type Repository struct { - Resources *Resources - Devices *Devices + Resources *Resources + Devices *Devices + Operations *Operations } func NewRepository(conn *sql.DB) *Repository { return &Repository{ - Resources: &Resources{DB: conn}, - Devices: &Devices{DB: conn}, + Resources: &Resources{DB: conn}, + Devices: &Devices{DB: conn}, + Operations: &Operations{DB: conn}, } } diff --git a/backend/repository/resources.go b/backend/repository/resources.go index 1b22760..77cec02 100644 --- a/backend/repository/resources.go +++ b/backend/repository/resources.go @@ -131,24 +131,20 @@ func (r *Resources) scanFile(scan func(...any) error) (FileRow, error) { return row, err } -// ListFiles returns the owner device's files (optionally within a folder), -// plus the total count matching the filter. +// ListFiles returns the owner device's files (within a folder, or at the +// root when folderResourceID is empty), plus the total count. func (r *Resources) ListFiles(ownerID, folderResourceID string, limit, offset int, sort, order string) ([]FileRow, int, error) { column, direction := sortClause(sort, order) - var folderFilter any - if folderResourceID != "" { - folderFilter = folderResourceID - } - where := `type = 'file' AND deleted_at IS NULL AND owner_id = $1 AND ($2::text IS NULL OR parent_id = $2)` + where := `type = 'file' AND deleted_at IS NULL AND owner_id = $1 AND ($2::text = '' AND parent_id IS NULL OR parent_id = $2)` var total int - if err := r.DB.QueryRow(`SELECT COUNT(*) FROM resources WHERE `+where, ownerID, folderFilter).Scan(&total); err != nil { + if err := r.DB.QueryRow(`SELECT COUNT(*) FROM resources WHERE `+where, ownerID, folderResourceID).Scan(&total); err != nil { return nil, 0, err } rows, err := r.DB.Query( fmt.Sprintf(`SELECT %s FROM resources WHERE %s ORDER BY %s %s LIMIT $3 OFFSET $4`, fileColumns, where, column, direction), - ownerID, folderFilter, limit, offset, + ownerID, folderResourceID, limit, offset, ) if err != nil { return nil, 0, err @@ -233,6 +229,133 @@ func escapeLike(q string) string { return replacer.Replace(q) } +// GetFolder returns a folder row owned by the device (no-rows → ErrNotFound). +func (r *Resources) GetFolder(ownerID, resourceID string) (FolderRow, error) { + var row FolderRow + err := r.DB.QueryRow( + `SELECT resource_id, name, COALESCE(parent_id, '') FROM resources + WHERE type = 'folder' AND deleted_at IS NULL AND owner_id = $1 AND resource_id = $2`, + ownerID, resourceID, + ).Scan(&row.ID, &row.Name, &row.ParentID) + if err == sql.ErrNoRows { + return FolderRow{}, ErrNotFound + } + return row, err +} + +// MoveResource sets parent_id (root when parentResourceID empty). The target +// folder must exist and belong to the owner. Absent source is a no-op. +func (r *Resources) MoveResource(ownerID, resourceID, parentResourceID string) error { + if parentResourceID != "" { + if _, err := r.GetFolder(ownerID, parentResourceID); err != nil { + return err + } + } + var parentID any + if parentResourceID != "" { + parentID = parentResourceID + } + result, err := r.DB.Exec( + `UPDATE resources SET parent_id = $3, updated_at = NOW() + WHERE resource_id = $1 AND owner_id = $2 AND deleted_at IS NULL`, + resourceID, ownerID, parentID, + ) + if err != nil && isUniqueViolation(err) { + return ErrNameConflict + } + if err != nil { + return err + } + affected, err := result.RowsAffected() + if err != nil { + return err + } + if affected == 0 { + return ErrNotFound + } + return nil +} + +// UpdateName renames a resource. Absent source is a no-op. +func (r *Resources) UpdateName(ownerID, resourceID, name string) error { + result, err := r.DB.Exec( + `UPDATE resources SET name = $3, updated_at = NOW() + WHERE resource_id = $1 AND owner_id = $2 AND deleted_at IS NULL`, + resourceID, ownerID, name, + ) + if err != nil && isUniqueViolation(err) { + return ErrNameConflict + } + if err != nil { + return err + } + affected, err := result.RowsAffected() + if err != nil { + return err + } + if affected == 0 { + return ErrNotFound + } + return nil +} + +// SyncDelete soft-deletes a resource; absence is NOT an error (idempotent +// terminal state for the outbox). +func (r *Resources) SyncDelete(ownerID, resourceID string) error { + _, err := r.DB.Exec( + `UPDATE resources SET deleted_at = NOW(), updated_at = NOW() + WHERE resource_id = $1 AND owner_id = $2 AND deleted_at IS NULL`, + ownerID, resourceID, + ) + return err +} + +// ExistsOwner reports whether a non-deleted resource belongs to the device. +func (r *Resources) ExistsOwner(ownerID, resourceID string) (bool, error) { + var exists int + err := r.DB.QueryRow( + `SELECT 1 FROM resources WHERE resource_id = $1 AND owner_id = $2 AND deleted_at IS NULL`, + resourceID, ownerID, + ).Scan(&exists) + if err == sql.ErrNoRows { + return false, nil + } + return err == nil, err +} + +// OwnedRow is a snapshot row: resource identity + freshness. +type OwnedRow struct { + ID string + Type string + UpdatedAt time.Time +} + +// ListOwned returns the device's non-deleted resources whose updated_at +// (in epoch ms) is strictly greater than afterMs (0 = all). +func (r *Resources) ListOwned(ownerID string, afterMs int64) ([]OwnedRow, error) { + rows, err := r.DB.Query( + `SELECT resource_id, type, updated_at FROM resources + WHERE owner_id = $1 AND deleted_at IS NULL + AND (EXTRACT(EPOCH FROM updated_at) * 1000)::bigint > $2 + ORDER BY updated_at DESC + LIMIT 10000`, + ownerID, afterMs, + ) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]OwnedRow, 0) + for rows.Next() { + var row OwnedRow + if err := rows.Scan(&row.ID, &row.Type, &row.UpdatedAt); err != nil { + return nil, err + } + out = append(out, row) + } + return out, rows.Err() +} + // ListRootFolders returns the owner device's top-level folders (parent_id NULL). func (r *Resources) ListRootFolders(ownerID string) ([]FolderRow, error) { rows, err := r.DB.Query( diff --git a/backend/service/sync.go b/backend/service/sync.go new file mode 100644 index 0000000..a153d06 --- /dev/null +++ b/backend/service/sync.go @@ -0,0 +1,257 @@ +package service + +import ( + "encoding/json" + "errors" + "regexp" + "time" + + "github.com/vaultdrop/backend/repository" +) + +var resourceIDPattern = regexp.MustCompile(`^[0-9a-f]{32}$`) + +// Ops de l'outbox (docs/api-v1.md §6.1). Les ops partage (share/share_link) +// sont accusées réception mais sans état serveur en V1 (single-owner). +const ( + OpCreateResource = "create_resource" + OpUpdateMetadata = "update_metadata" + OpDeleteResource = "delete_resource" + OpMoveResource = "move_resource" + OpShare = "share" + OpRevokeShare = "revoke_share" + OpUpdateShare = "update_share" + OpCreateLink = "create_link" + OpRevokeLink = "revoke_link" +) + +var ackOnlyOps = map[string]bool{ + OpShare: true, OpRevokeShare: true, OpUpdateShare: true, + OpCreateLink: true, OpRevokeLink: true, +} + +// SyncOperation is one outbox entry (shadow of mobile PendingOperationRow). +type SyncOperation struct { + OperationID int64 `json:"operation_id"` + RefType string `json:"ref_type"` + RefID int64 `json:"ref_id"` + ResourceID string `json:"resource_id"` + ResourceType string `json:"resource_type"` + Operation string `json:"operation"` + Payload json.RawMessage `json:"payload"` +} + +// FailedOperation reports the first non-idempotent failure. +type FailedOperation struct { + OperationID int64 `json:"operation_id"` + Code string `json:"code"` + Message string `json:"message"` +} + +// SyncResult is the outbox response: applied = index of the next op to send. +type SyncResult struct { + Applied int `json:"applied"` + Failed *FailedOperation `json:"failed,omitempty"` +} + +type createResourcePayload struct { + Name string `json:"name"` + MimeType string `json:"mimeType"` + Extension string `json:"extension"` +} + +type renamePayload struct { + Name string `json:"name"` +} + +type movePayload struct { + ToFolderResourceID string `json:"toFolderResourceId"` +} + +func classifySyncError(err error) (string, string) { + switch { + case errors.Is(err, repository.ErrNameConflict): + return "NAME_CONFLICT", "a resource with this name already exists here" + case errors.Is(err, repository.ErrNotFound): + return "NOT_FOUND", "target resource or folder not found" + default: + return "INVALID_REQUEST", err.Error() + } +} + +func (s *Resources) ApplyBatch(ownerID string, ops []SyncOperation) (SyncResult, error) { + for i := range ops { + op := &ops[i] + if err := validateSyncOp(op); err != nil { + return SyncResult{Applied: i, Failed: &FailedOperation{OperationID: op.OperationID, Code: "INVALID_REQUEST", Message: err.Error()}}, nil + } + already, err := s.Repository.Operations.Applied(ownerID, op.OperationID) + if err != nil { + return SyncResult{}, err + } + if already { + continue + } + if ackOnlyOps[op.Operation] { + if err := s.recordApplied(ownerID, op); err != nil { + return SyncResult{}, err + } + continue + } + if err := s.applySyncOp(ownerID, op); err != nil { + code, message := classifySyncError(err) + return SyncResult{Applied: i, Failed: &FailedOperation{OperationID: op.OperationID, Code: code, Message: message}}, nil + } + if err := s.recordApplied(ownerID, op); err != nil { + return SyncResult{}, err + } + } + return SyncResult{Applied: len(ops)}, nil +} + +func validateSyncOp(op *SyncOperation) error { + if op.OperationID <= 0 { + return errors.New("operation_id must be > 0") + } + if op.Operation == "" { + return errors.New("missing operation type") + } + if ackOnlyOps[op.Operation] { + return nil + } + if !resourceIDPattern.MatchString(op.ResourceID) { + return errors.New("resource_id must be 32 lowercase hex chars") + } + switch op.Operation { + case OpCreateResource, OpUpdateMetadata, OpMoveResource, OpDeleteResource: + default: + return errors.New("unknown operation " + op.Operation) + } + if op.ResourceType != "folder" && op.ResourceType != "file" { + return errors.New("resource_type must be 'folder' or 'file'") + } + return nil +} + +func (s *Resources) applySyncOp(ownerID string, op *SyncOperation) error { + switch op.Operation { + case OpCreateResource: + exists, err := s.Repo.ExistsOwner(ownerID, op.ResourceID) + if err != nil { + return err + } + if exists { + return nil + } + var p createResourcePayload + if err := json.Unmarshal(op.Payload, &p); err != nil { + return errors.New("invalid payload: " + err.Error()) + } + if p.Name == "" { + return errors.New("payload.name required") + } + if op.ResourceType == "folder" { + return s.Repo.InsertFolder(ownerID, op.ResourceID, p.Name, "") + } + mime := p.MimeType + return s.Repo.InsertFile(ownerID, op.ResourceID, p.Name, "", 0, &mime, nullableString(p.Extension)) + + case OpUpdateMetadata: + exists, err := s.Repo.ExistsOwner(ownerID, op.ResourceID) + if err != nil { + return err + } + if !exists { + return nil + } + var p renamePayload + if err := json.Unmarshal(op.Payload, &p); err != nil { + return errors.New("invalid payload: " + err.Error()) + } + if p.Name == "" { + return errors.New("payload.name required") + } + return s.Repo.UpdateName(ownerID, op.ResourceID, p.Name) + + case OpMoveResource: + exists, err := s.Repo.ExistsOwner(ownerID, op.ResourceID) + if err != nil { + return err + } + if !exists { + return nil + } + var p movePayload + if err := json.Unmarshal(op.Payload, &p); err != nil { + return errors.New("invalid payload: " + err.Error()) + } + if p.ToFolderResourceID == op.ResourceID { + return errors.New("cannot move a resource into itself") + } + return s.Repo.MoveResource(ownerID, op.ResourceID, p.ToFolderResourceID) + + case OpDeleteResource: + return s.Repo.SyncDelete(ownerID, op.ResourceID) + + default: + return errors.New("unknown operation " + op.Operation) + } +} + +func (s *Resources) recordApplied(ownerID string, op *SyncOperation) error { + return s.Repository.Operations.Record(ownerID, op.OperationID, op.Operation, op.RefType, nullableInt64(op.RefID), op.ResourceID, op.Payload) +} + +func nullableString(value string) *string { + if value == "" { + return nil + } + return &value +} + +func nullableInt64(value int64) *int64 { + if value == 0 { + return nil + } + return &value +} + +// ResourcePermission is the snapshot shape consumed by canAccess +// (docs/api-v1.md §6.2). +type ResourcePermission struct { + ResourceID string `json:"resource_id"` + ResourceType string `json:"resourceType"` + EffectiveAccess string `json:"effectiveAccess"` + Inherit bool `json:"inherit"` + OwnerID string `json:"ownerId"` + SharedByID any `json:"sharedById"` + ExpiresAt any `json:"expiresAt"` + CachedAt int64 `json:"cachedAt"` + UpdatedAt int64 `json:"updatedAt"` +} + +// Snapshot returns the delta of effective permissions for the device since +// afterMs (epoch ms; 0 = all). V1 single-owner : toutes les ressources +// appartiennent au device appelant (effective_access = owner). +func (s *Resources) Snapshot(ownerID string, afterMs int64) ([]ResourcePermission, error) { + rows, err := s.Repo.ListOwned(ownerID, afterMs) + if err != nil { + return nil, err + } + now := time.Now().UnixMilli() + perms := make([]ResourcePermission, 0, len(rows)) + for _, row := range rows { + perms = append(perms, ResourcePermission{ + ResourceID: row.ID, + ResourceType: row.Type, + EffectiveAccess: "owner", + Inherit: false, + OwnerID: ownerID, + SharedByID: nil, + ExpiresAt: nil, + CachedAt: now, + UpdatedAt: row.UpdatedAt.UnixMilli(), + }) + } + return perms, nil +} diff --git a/docs/api-v1.md b/docs/api-v1.md index d1f953a..c14107e 100644 --- a/docs/api-v1.md +++ b/docs/api-v1.md @@ -93,7 +93,13 @@ type OcrJob = { id: string; status: OcrJobStatus; text?: string | null; error?: - **Idempotence** : contrainte d'unicité serveur `(device_id, operation_id)`. Pour chaque op : si déjà traitée → **no-op** (comptée comme appliquée, les doublons arrivent à cause du backoff/retry). Sinon appliquée si valide. - **Ordre** : les opérations sont appliquées **séquentiellement**, dans l'ordre du batch. Le serveur **s'arrête à la première erreur non-idempotente** et renvoie l'index atteint — le client reprend à cet index. - Réponse : `2xx` avec `{ "applied": int, "failed": { "operation_id": int, "code": string, "message": string } | null }` (`applied` = index de la prochaine op à envoyer). -- Côté client, le `pushStatus` (pending/synced/failed) des shares/share_links est **dérivé** de l'état des opérations de l'outbox ; dead-letter après `MAX_PENDING_ATTEMPTS` (= 5). +- Côté client, le `pushStatus` (pending/synced/failed) des shares/share_links est **dérivé** de l'état des opérations de l'outbox ; dead-letter après `MAX_PENDING_ATTEMPTS` (= 5). **Côté serveur, les ops `share | revoke_share | update_share | create_link | revoke_link` sont accusées réception mais ne créent aucun état** (V1 single-owner, pas de table shares serveur) — la dérivation du pushStatus reste purement client. +- Sémantique d'application (côté serveur) : + - `create_resource` : crée la ressource ; **déjà présente → no-op** (rejeu idempotent). `payload.name` obligatoire. + - `update_metadata` / `move_resource` : ressource absente → **no-op** (état terminal atteint) ; dossier cible de `move_resource` absent → `NOT_FOUND` ; déplacement dans soi-même → `INVALID_REQUEST`. + - `delete_resource` : **idempotent** — suppression d'une ressource absente = succès. + - Validation (deuxième champ `operation_id`, hex32 pour `resource_id`, enum `operation`) → échec `INVALID_REQUEST` avec arrêt du batch. + - Nom déjà pris (même parent, ou à la racine) → échec `NAME_CONFLICT`.** ### 6.2 Snapshot — `GET /sync/permissions?after=` @@ -127,4 +133,4 @@ type ResourcePermission = { ## 7. Codes d'erreur courants -`NOT_FOUND`, `NOT_IMPLEMENTED` (501 temporaire sur les routes non construites — état actuel : files CRUD/upload/search, folders, devices, health sont réels ; `ocr/*`, `sync/*` en queue), `FILE_TOO_LARGE` (413), `NAME_CONFLICT` (409 — même nom dans le même parent, cf. `UNIQUE(parent_id, name)`), `NETWORK_ERROR` (côté client), `HTTP_` (fallback). Le serveur doit répondre 501 `{ "error": { "code": "NOT_IMPLEMENTED", "message": "…" } }` sur toute route encore en queue. Statut `SERVICE_UNAVAILABLE` (503) si le backend n'est pas initialisé. \ No newline at end of file +`NOT_FOUND`, `NOT_IMPLEMENTED` (501 temporaire sur les routes non construites — état actuel : files CRUD/upload/search, folders, devices, health, **sync/ops + sync/permissions** sont réels ; `ocr/*` en queue), `FILE_TOO_LARGE` (413), `NAME_CONFLICT` (409 — même nom dans le même parent, cf. `UNIQUE(parent_id, name)`, **ou à la racine**, index partiel `(owner_id, name) WHERE parent_id IS NULL`), `NETWORK_ERROR` (côté client), `HTTP_` (fallback). Le serveur doit répondre 501 `{ "error": { "code": "NOT_IMPLEMENTED", "message": "…" } }` sur toute route encore en queue. Statut `SERVICE_UNAVAILABLE` (503) si le backend n'est pas initialisé. \ No newline at end of file