Implement Disklet sync

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 ## Unreleased
 
+- added: `syncRepo(disklet, syncKey, lastHash)` method on `SyncClient` for Disklet-based repository synchronization with `changes/`, `deleted/`, and `data/` staging directories
+
+## 0.2.9-1 (2024-07-30)
+
+- removed: Remove InfoClient usage for SyncClient instances
+
 ## 0.2.9 (2024-02-26)
 
 - added: Detect conflicts while creating repos, and report these with a new `ConflictError` type.

package.json

@@ -1,6 +1,6 @@
 {
   "name": "edge-sync-client",
-  "version": "0.2.9",
+  "version": "0.2.9-1",
   "description": "Library for accessing the Edge data-sync system",
   "keywords": [
     "Edge"
@@ -59,6 +59,9 @@
     "rfc4648": "^1.5.0",
     "serverlet": "^0.1.0"
   },
+  "peerDependencies": {
+    "disklet": ">=0.5.0"
+  },
   "devDependencies": {
     "@babel/core": "^7.10.4",
     "@rollup/plugin-node-resolve": "^8.1.0",
@@ -71,6 +74,7 @@
     "@typescript-eslint/parser": "^4.8.2",
     "babel-eslint": "^10.1.0",
     "chai": "^4.2.0",
+    "disklet": "^0.5.2",
     "eslint": "^7.14.0",
     "eslint-config-standard-kit": "0.15.1",
     "eslint-plugin-flowtype": "^5.2.0",

rollup.config.js

@@ -8,7 +8,11 @@ import packageJson from './package.json'
 const extensions = ['.ts']
 
 export default {
-  external: [/@babel\/runtime/, ...Object.keys(packageJson.dependencies)],
+  external: [
+    /@babel\/runtime/,
+    ...Object.keys(packageJson.dependencies),
+    ...Object.keys(packageJson.peerDependencies || {})
+  ],
   input: 'src/index.ts',
   output: [
     { file: packageJson.main, format: 'cjs' },

src/client/sync-client.ts

@@ -1,23 +1,37 @@
 import { asMaybe, Cleaner, uncleaner } from 'cleaners'
 import crossFetch from 'cross-fetch'
+import { Disklet, navigateDisklet } from 'disklet'
 import { FetchFunction, FetchResponse } from 'serverlet'
 
-import { EdgeServers } from '../types/base-types'
+import { asEdgeBox, EdgeServers, wasEdgeBox } from '../types/base-types'
 import { ConflictError } from '../types/error'
 import {
   asGetStoreResponse,
   asPostStoreBody,
   asPostStoreResponse,
   asPutStoreResponse,
   asServerErrorResponse,
+  ChangeSet,
   GetStoreResponse,
   PostStoreBody,
   PostStoreResponse,
   PutStoreResponse
 } from '../types/rest-types'
 import { syncKeyToRepoId } from '../util/security'
 import { shuffle } from '../util/shuffle'
-import { makeInfoClient } from './info-client'
+
+const defaultEdgeServers: Required<EdgeServers> = {
+  infoServers: ['https://info-eu1.edge.app', 'https://info-us1.edge.app'],
+  syncServers: [
+    'https://sync-us1.edge.app',
+    'https://sync-us2.edge.app',
+    'https://sync-us3.edge.app',
+    'https://sync-us4.edge.app',
+    'https://sync-us5.edge.app',
+    'https://sync-us6.edge.app',
+    'https://sync-eu.edge.app'
+  ]
+}
 
 export interface SyncClient {
   createRepo: (syncKey: string, apiKey?: string) => Promise<PutStoreResponse>
@@ -30,21 +44,47 @@ export interface SyncClient {
     lastHash: string | undefined,
     body: PostStoreBody
   ) => Promise<PostStoreResponse>
+  /**
+   * Sync the repository using the provided Disklet.
+   * Gathers changes from `changes/` and deletions from `deleted/` directories,
+   * sends them to the server, applies server's response to `data/`, and clears staging directories.
+   */
+  syncRepo: (
+    disklet: Disklet,
+    syncKey: string,
+    lastHash: string | undefined
+  ) => Promise<SyncResult>
+}
+
+export interface SyncResult {
+  /** The sync status after sync */
+  status: SyncStatus
+  /** The changes received from the server */
+  changes: ChangeSet
+}
+
+export interface SyncStatus {
+  /** The last known hash from the server */
+  lastHash: string | undefined
+  /** Unix timestamp of the last sync (seconds) */
+  lastSync: number
 }
 
 export interface SyncClientOptions {
   fetch?: FetchFunction
   log?: (message: string) => void
   edgeServers?: EdgeServers
+  /** Maximum number of changes to send per sync (default: 100) */
+  maxChangesPerSync?: number
 }
 
 export function makeSyncClient(opts: SyncClientOptions = {}): SyncClient {
-  const { fetch = crossFetch, log = () => {} } = opts
-  const infoClient = makeInfoClient(opts)
+  const { fetch = crossFetch, log = () => {}, maxChangesPerSync = 100 } = opts
+  const syncServers: Required<EdgeServers>['syncServers'] =
+    opts.edgeServers?.syncServers ?? defaultEdgeServers.syncServers
 
   // Returns the sync servers from the info client shuffled
   async function shuffledSyncServers(): Promise<string[]> {
-    const { syncServers } = await infoClient.getEdgeServers()
     return shuffle(syncServers)
   }
 
@@ -170,6 +210,94 @@ export function makeSyncClient(opts: SyncClientOptions = {}): SyncClient {
       }
 
       throw error
+    },
+
+    async syncRepo(disklet, syncKey, lastHash) {
+      // Get subdisklets for changes, deletions, and data
+      const changesDisklet = navigateDisklet(disklet, 'changes')
+      const deletedDisklet = navigateDisklet(disklet, 'deleted')
+      const dataDisklet = navigateDisklet(disklet, 'data')
+
+      // List both directories (each with limit to avoid over-listing)
+      const allChangePaths = await deepListWithLimit(
+        changesDisklet,
+        maxChangesPerSync
+      )
+      const allDeletePaths = await deepListWithLimit(
+        deletedDisklet,
+        maxChangesPerSync
+      )
+
+      // Interlace changes and deletions, respecting the limit
+      const changePaths: string[] = []
+      const deletePaths: string[] = []
+      const outgoingChanges: ChangeSet = {}
+      const maxChangesCount = Math.min(
+        maxChangesPerSync,
+        allChangePaths.length + allDeletePaths.length
+      )
+      for (let i = 0; i < maxChangesCount; i++) {
+        const pickChange = async (): Promise<void> => {
+          const path = allChangePaths[changePaths.length]
+          const data = await changesDisklet.getText(path)
+          outgoingChanges[path] = asEdgeBox(JSON.parse(data))
+          changePaths.push(path)
+        }
+        const pickDeletion = (): void => {
+          const path = allDeletePaths[deletePaths.length]
+          outgoingChanges[path] = null
+          deletePaths.push(path)
+        }
+
+        if (i % 2 === 0) {
+          if (changePaths.length < allChangePaths.length) {
+            await pickChange()
+          } else {
+            pickDeletion()
+          }
+        } else {
+          if (deletePaths.length < allDeletePaths.length) {
+            pickDeletion()
+          } else {
+            await pickChange()
+          }
+        }
+      }
+
+      // Use readRepo if no changes, updateRepo otherwise
+      const hasChanges = Object.keys(outgoingChanges).length > 0
+      const response = hasChanges
+        ? await this.updateRepo(syncKey, lastHash, { changes: outgoingChanges })
+        : await this.readRepo(syncKey, lastHash)
+
+      // Apply server's changes to the data disklet
+      for (const [path, change] of Object.entries(response.changes)) {
+        if (change === null) {
+          // Delete the file from data directory
+          await dataDisklet.delete(path)
+        } else {
+          // Write the file to data directory
+          await dataDisklet.setText(path, JSON.stringify(wasEdgeBox(change)))
+        }
+      }
+
+      // Clear synced changes from changes/ directory
+      for (const path of changePaths) {
+        await changesDisklet.delete(path)
+      }
+
+      // Clear synced deletions from deleted/ directory
+      for (const path of deletePaths) {
+        await deletedDisklet.delete(path)
+      }
+
+      return {
+        status: {
+          lastHash: response.hash ?? lastHash,
+          lastSync: Date.now() / 1000
+        },
+        changes: response.changes
+      }
     }
   }
 }
@@ -183,3 +311,30 @@ interface ApiRequest {
 }
 
 const wasPostStoreBody = uncleaner(asPostStoreBody)
+
+// Disklet helper functions
+
+/**
+ * Lists all files in a disklet recursively, up to a limit.
+ * Returns a list of full paths.
+ */
+
+async function deepListWithLimit(
+  disklet: Disklet,
+  limit: number,
+  path: string = ''
+): Promise<string[]> {
+  const list = await disklet.list(path)
+  const paths = Object.keys(list).filter(path => list[path] === 'file')
+  const folders = Object.keys(list).filter(path => list[path] === 'folder')
+
+  // Loop over folders to get subpaths
+  for (const folder of folders) {
+    if (paths.length >= limit) break
+    const remaining = limit - paths.length
+    const subpaths = await deepListWithLimit(disklet, remaining, folder)
+    paths.push(...subpaths.slice(0, remaining))
+  }
+
+  return paths
+}

src/index.flow.js

@@ -1,48 +1,58 @@
 // @flow
 
+import { type Disklet } from 'disklet'
 import { type FetchFunction } from 'serverlet'
 
 type EdgeServers = {
   infoServers?: string[],
   syncServers?: string[]
 }
 
-type SyncClientOptions = {
-  fetch?: FetchFunction,
-  log?: (message: string) => void,
-  edgeServers?: EdgeServers
-}
-
 type EdgeBox = {
-  iv_hex: string,
   encryptionType: number,
-  data_base64: string
+  data_base64: string,
+  iv_hex: string
+}
+
+type ChangeSet = {
+  [path: string]: EdgeBox | null
 }
 
 type PutStoreResponse = void
 
 type GetStoreResponse = {
   hash?: string | void,
-  changes: {
-    [keys: string]: EdgeBox | null
-  }
+  changes: ChangeSet
 }
 
 type PostStoreBody = {
-  changes: {
-    [keys: string]: EdgeBox | null
-  }
+  changes: ChangeSet
 }
 
 type PostStoreResponse = {
   hash: string,
-  changes: {
-    [keys: string]: EdgeBox | null
-  }
+  changes: ChangeSet
+}
+
+export type SyncStatus = {
+  lastHash: string | void,
+  lastSync: number
+}
+
+export type SyncResult = {
+  status: SyncStatus,
+  changes: ChangeSet
+}
+
+export type SyncClientOptions = {
+  fetch?: FetchFunction,
+  log?: (message: string) => void,
+  edgeServers?: EdgeServers,
+  maxChangesPerSync?: number
 }
 
 export type SyncClient = {
-  createRepo: (syncKey: string) => Promise<PutStoreResponse>,
+  createRepo: (syncKey: string, apiKey?: string) => Promise<PutStoreResponse>,
   readRepo: (
     syncKey: string,
     lastHash: string | void
@@ -51,7 +61,12 @@ export type SyncClient = {
     syncKey: string,
     lastHash: string | void,
     body: PostStoreBody
-  ) => Promise<PostStoreResponse>
+  ) => Promise<PostStoreResponse>,
+  syncRepo: (
+    disklet: Disklet,
+    syncKey: string,
+    lastHash: string | void
+  ) => Promise<SyncResult>
 }
 
-declare export function makeSyncClient(opts: SyncClientOptions): SyncClient
+declare export function makeSyncClient(opts?: SyncClientOptions): SyncClient

src/index.ts

@@ -9,7 +9,12 @@ export * from './types/rest-types'
 export { asMaybeConflictError, ConflictError } from './types/error'
 
 // Client
-export type { SyncClient, SyncClientOptions } from './client/sync-client'
+export type {
+  SyncStatus,
+  SyncClient,
+  SyncClientOptions,
+  SyncResult
+} from './client/sync-client'
 export { makeSyncClient } from './client/sync-client'
 
 // Util