Initial commit

apps/sync-server/src/index.ts

@@ -0,0 +1,38 @@
+import path from "node:path";
+
+import cors from "cors";
+import express from "express";
+
+import { createJsonFileLogger } from "./logging/jsonFileLogger";
+import { createRequestLoggingMiddleware } from "./middleware/requestLoggingMiddleware";
+import { createSyncRouter } from "./routes/createSyncRouter";
+import { createFileSyncStore } from "./store/fileSyncStore";
+
+const port = Number(process.env.PORT ?? 8787);
+const dataDirectory = process.env.SYNC_DATA_DIR
+  ? path.resolve(process.env.SYNC_DATA_DIR)
+  : path.resolve(process.cwd(), "data");
+const app = express();
+const logger = createJsonFileLogger(path.join(dataDirectory, "logs", "server.jsonl"));
+const store = createFileSyncStore(path.join(dataDirectory, "sync-state.json"));
+
+app.use(cors());
+app.use(express.json({ limit: "10mb" }));
+app.use(createRequestLoggingMiddleware(logger));
+
+app.get("/health", (_request, response) => {
+  response.json({
+    status: "ok",
+    dataDirectory
+  });
+});
+
+app.use("/api", createSyncRouter(store, logger));
+
+app.listen(port, () => {
+  logger.info("server-started", {
+    port,
+    dataDirectory
+  });
+  console.log(`Obsidian Sync server listening on http://localhost:${port}`);
+});

apps/sync-server/src/logging/jsonFileLogger.ts

@@ -0,0 +1,45 @@
+import { appendFileSync, mkdirSync } from "node:fs";
+import { dirname } from "node:path";
+
+export type LogLevel = "debug" | "info" | "warn" | "error";
+
+export interface StructuredLogger {
+  log(level: LogLevel, message: string, context?: Record<string, unknown>): void;
+  debug(message: string, context?: Record<string, unknown>): void;
+  info(message: string, context?: Record<string, unknown>): void;
+  warn(message: string, context?: Record<string, unknown>): void;
+  error(message: string, context?: Record<string, unknown>): void;
+}
+
+export function createJsonFileLogger(logFilePath: string): StructuredLogger {
+  mkdirSync(dirname(logFilePath), { recursive: true });
+
+  const write = (level: LogLevel, message: string, context?: Record<string, unknown>) => {
+    const payload = {
+      timestamp: new Date().toISOString(),
+      level,
+      message,
+      context: context ?? {}
+    };
+
+    appendFileSync(logFilePath, `${JSON.stringify(payload)}\n`, "utf8");
+  };
+
+  return {
+    log(level, message, context) {
+      write(level, message, context);
+    },
+    debug(message, context) {
+      write("debug", message, context);
+    },
+    info(message, context) {
+      write("info", message, context);
+    },
+    warn(message, context) {
+      write("warn", message, context);
+    },
+    error(message, context) {
+      write("error", message, context);
+    }
+  };
+}

apps/sync-server/src/middleware/requestLoggingMiddleware.ts

@@ -0,0 +1,26 @@
+import type { Request, Response } from "express";
+import type { NextFunction } from "express";
+
+import type { StructuredLogger } from "../logging/jsonFileLogger";
+
+export function createRequestLoggingMiddleware(logger: StructuredLogger) {
+  return (request: Request, response: Response, next: NextFunction) => {
+    const requestId = globalThis.crypto.randomUUID();
+    const startedAt = Date.now();
+
+    response.locals.requestId = requestId;
+    response.setHeader("x-request-id", requestId);
+
+    response.on("finish", () => {
+      logger.info("http-request", {
+        requestId,
+        method: request.method,
+        path: request.originalUrl,
+        statusCode: response.statusCode,
+        durationMs: Date.now() - startedAt
+      });
+    });
+
+    next();
+  };
+}

apps/sync-server/src/routes/createSyncRouter.ts

@@ -0,0 +1,206 @@
+import type { Request, Response } from "express";
+import express from "express";
+
+import {
+  type ClientLogUploadRequest,
+  type ListDevicesRequest,
+  type RegisterDeviceRequest,
+  type RevokeDeviceRequest,
+  type RotateVaultKeyRequest,
+  type SyncPullRequest,
+  type SyncPushRequest,
+  ClientLogUploadRequestSchema,
+  ClientLogUploadResponseSchema,
+  ListDevicesRequestSchema,
+  ListDevicesResponseSchema,
+  RegisterDeviceRequestSchema,
+  RevokeDeviceRequestSchema,
+  RevokeDeviceResponseSchema,
+  RotateVaultKeyRequestSchema,
+  RotateVaultKeyResponseSchema,
+  SyncPullRequestSchema,
+  SyncPushRequestSchema,
+  parseWithSchema
+} from "@obsidian-sync/sync-protocol";
+
+import type { StructuredLogger } from "../logging/jsonFileLogger";
+import type { SyncStore } from "../store/syncStore";
+
+function readBearerToken(request: Request): string | undefined {
+  const authorization = request.header("authorization");
+  return authorization?.replace(/^Bearer\s+/i, "");
+}
+
+function getRequestId(response: Response): string {
+  return String(response.locals.requestId ?? "unknown-request");
+}
+
+function unauthorized(request: Request, response: Response, logger: StructuredLogger): void {
+  logger.warn("request-unauthorized", {
+    requestId: getRequestId(response),
+    method: request.method,
+    path: request.originalUrl
+  });
+
+  response.status(401).json({
+    error: "Unauthorized"
+  });
+}
+
+function badRequest(request: Request, response: Response, logger: StructuredLogger, error: unknown): void {
+  logger.warn("request-invalid", {
+    requestId: getRequestId(response),
+    method: request.method,
+    path: request.originalUrl,
+    error: error instanceof Error ? error.message : "Invalid request"
+  });
+
+  response.status(400).json({
+    error: error instanceof Error ? error.message : "Invalid request"
+  });
+}
+
+export function createSyncRouter(store: SyncStore, logger: StructuredLogger): express.Router {
+  const router = express.Router();
+
+  router.post("/devices/register", (request: Request, response: Response) => {
+    try {
+      const body: RegisterDeviceRequest = parseWithSchema(RegisterDeviceRequestSchema, request.body);
+      const registered = store.registerDevice(body);
+      logger.info("device-registered", {
+        requestId: getRequestId(response),
+        vaultId: registered.vaultId,
+        deviceId: registered.deviceId,
+        deviceName: body.deviceName
+      });
+      response.status(201).json(registered);
+    } catch (error) {
+      badRequest(request, response, logger, error);
+    }
+  });
+
+  router.post("/sync/pull", (request: Request, response: Response) => {
+    try {
+      const body: SyncPullRequest = parseWithSchema(SyncPullRequestSchema, request.body);
+      const token = readBearerToken(request);
+      if (!token || !store.authenticate(body.vaultId, body.deviceId, token)) {
+        unauthorized(request, response, logger);
+        return;
+      }
+
+      response.json(store.pull(body.vaultId, body.sinceServerRevision, body.limit));
+    } catch (error) {
+      badRequest(request, response, logger, error);
+    }
+  });
+
+  router.post("/sync/push", (request: Request, response: Response) => {
+    try {
+      const body: SyncPushRequest = parseWithSchema(SyncPushRequestSchema, request.body);
+      const token = readBearerToken(request);
+      if (!token || !store.authenticate(body.vaultId, body.deviceId, token)) {
+        unauthorized(request, response, logger);
+        return;
+      }
+
+      response.json(store.push(body));
+    } catch (error) {
+      badRequest(request, response, logger, error);
+    }
+  });
+
+  router.post("/devices/list", (request: Request, response: Response) => {
+    try {
+      const body: ListDevicesRequest = parseWithSchema(ListDevicesRequestSchema, request.body);
+      const token = readBearerToken(request);
+      if (!token || !store.authenticate(body.vaultId, body.deviceId, token)) {
+        unauthorized(request, response, logger);
+        return;
+      }
+
+      response.json(ListDevicesResponseSchema.parse(store.listDevices(body.vaultId)));
+    } catch (error) {
+      badRequest(request, response, logger, error);
+    }
+  });
+
+  router.post("/devices/revoke", (request: Request, response: Response) => {
+    try {
+      const body: RevokeDeviceRequest = parseWithSchema(RevokeDeviceRequestSchema, request.body);
+      const token = readBearerToken(request);
+      if (!token || !store.authenticate(body.vaultId, body.deviceId, token)) {
+        unauthorized(request, response, logger);
+        return;
+      }
+
+      const revoked = store.revokeDevice(body.vaultId, body.targetDeviceId);
+      logger.warn("device-revoked", {
+        requestId: getRequestId(response),
+        vaultId: body.vaultId,
+        requestedBy: body.deviceId,
+        targetDeviceId: body.targetDeviceId,
+        revokedAt: revoked.revokedAt
+      });
+      response.json(RevokeDeviceResponseSchema.parse(revoked));
+    } catch (error) {
+      badRequest(request, response, logger, error);
+    }
+  });
+
+  router.post("/keys/rotate", (request: Request, response: Response) => {
+    try {
+      const body: RotateVaultKeyRequest = parseWithSchema(RotateVaultKeyRequestSchema, request.body);
+      const token = readBearerToken(request);
+      if (!token || !store.authenticate(body.vaultId, body.deviceId, token)) {
+        unauthorized(request, response, logger);
+        return;
+      }
+
+      const rotated = store.rotateVaultKey(body.vaultId, body.nextKeyId, body.previousKeyId);
+      logger.info("vault-key-rotated", {
+        requestId: getRequestId(response),
+        vaultId: body.vaultId,
+        requestedBy: body.deviceId,
+        activeKeyId: rotated.activeKeyId,
+        rotatedAt: rotated.rotatedAt
+      });
+      response.json(RotateVaultKeyResponseSchema.parse(rotated));
+    } catch (error) {
+      badRequest(request, response, logger, error);
+    }
+  });
+
+  router.post("/logs", (request: Request, response: Response) => {
+    try {
+      const body: ClientLogUploadRequest = parseWithSchema(ClientLogUploadRequestSchema, request.body);
+      const token = readBearerToken(request);
+      if (!token || !store.authenticate(body.vaultId, body.deviceId, token)) {
+        unauthorized(request, response, logger);
+        return;
+      }
+
+      for (const entry of body.entries) {
+        logger.log(entry.level, "client-sync-log", {
+          requestId: getRequestId(response),
+          vaultId: body.vaultId,
+          deviceId: body.deviceId,
+          runId: body.runId,
+          clientTimestamp: entry.timestamp,
+          clientMessage: entry.message,
+          context: entry.context ?? {}
+        });
+      }
+
+      response.status(202).json(
+        ClientLogUploadResponseSchema.parse({
+          accepted: body.entries.length,
+          requestId: getRequestId(response)
+        })
+      );
+    } catch (error) {
+      badRequest(request, response, logger, error);
+    }
+  });
+
+  return router;
+}

apps/sync-server/src/store/fileSyncStore.ts

@@ -0,0 +1,341 @@
+import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs";
+import { dirname } from "node:path";
+
+import type {
+  ListDevicesResponse,
+  RegisterDeviceRequest,
+  RegisterDeviceResponse,
+  RevokeDeviceResponse,
+  RotateVaultKeyResponse,
+  SyncChange,
+  SyncConflict,
+  SyncFileRecord,
+  SyncPullResponse,
+  SyncPushRequest,
+  SyncPushResponse,
+  Tombstone
+} from "@obsidian-sync/sync-protocol";
+
+import type { SyncStore } from "./syncStore";
+
+interface VaultDevice {
+  deviceId: string;
+  deviceName: string;
+  token: string;
+  issuedAt: string;
+  revokedAt?: string;
+}
+
+interface VaultState {
+  serverRevision: number;
+  devices: Map<string, VaultDevice>;
+  files: Map<string, SyncFileRecord>;
+  tombstones: Map<string, Tombstone>;
+  changes: SyncChange[];
+  activeKeyId?: string;
+  keyRotatedAt?: string;
+}
+
+interface PersistedVaultState {
+  serverRevision: number;
+  devices: Record<string, VaultDevice>;
+  files: Record<string, SyncFileRecord>;
+  tombstones: Record<string, Tombstone>;
+  changes: SyncChange[];
+  activeKeyId?: string;
+  keyRotatedAt?: string;
+}
+
+interface PersistedStore {
+  vaults: Record<string, PersistedVaultState>;
+}
+
+function createVaultState(): VaultState {
+  return {
+    serverRevision: 0,
+    devices: new Map(),
+    files: new Map(),
+    tombstones: new Map(),
+    changes: []
+  };
+}
+
+function mapFromRecord<TValue>(record: Record<string, TValue> | undefined): Map<string, TValue> {
+  return new Map(Object.entries(record ?? {}));
+}
+
+function recordFromMap<TValue>(map: Map<string, TValue>): Record<string, TValue> {
+  return Object.fromEntries(map.entries());
+}
+
+function loadVaults(filePath: string): Map<string, VaultState> {
+  if (!existsSync(filePath)) {
+    return new Map();
+  }
+
+  const raw = readFileSync(filePath, "utf8");
+  if (!raw.trim()) {
+    return new Map();
+  }
+
+  const parsed = JSON.parse(raw) as Partial<PersistedStore>;
+  const vaults = new Map<string, VaultState>();
+
+  for (const [vaultId, vault] of Object.entries(parsed.vaults ?? {})) {
+    vaults.set(vaultId, {
+      serverRevision: vault.serverRevision ?? 0,
+      devices: mapFromRecord(vault.devices),
+      files: mapFromRecord(vault.files),
+      tombstones: mapFromRecord(vault.tombstones),
+      changes: [...(vault.changes ?? [])],
+      activeKeyId: vault.activeKeyId,
+      keyRotatedAt: vault.keyRotatedAt
+    });
+  }
+
+  return vaults;
+}
+
+function persistVaults(filePath: string, vaults: Map<string, VaultState>): void {
+  mkdirSync(dirname(filePath), { recursive: true });
+
+  const payload: PersistedStore = {
+    vaults: Object.fromEntries(
+      [...vaults.entries()].map(([vaultId, vault]) => [
+        vaultId,
+        {
+          serverRevision: vault.serverRevision,
+          devices: recordFromMap(vault.devices),
+          files: recordFromMap(vault.files),
+          tombstones: recordFromMap(vault.tombstones),
+          changes: vault.changes,
+          activeKeyId: vault.activeKeyId,
+          keyRotatedAt: vault.keyRotatedAt
+        }
+      ])
+    )
+  };
+
+  const temporaryPath = `${filePath}.tmp`;
+  writeFileSync(temporaryPath, JSON.stringify(payload, null, 2), "utf8");
+  renameSync(temporaryPath, filePath);
+}
+
+function appendFileChange(vault: VaultState, file: SyncFileRecord): void {
+  vault.serverRevision += 1;
+  vault.changes.push({
+    serverRevision: vault.serverRevision,
+    file
+  });
+}
+
+function appendTombstoneChange(vault: VaultState, tombstone: Tombstone): void {
+  vault.serverRevision += 1;
+  vault.changes.push({
+    serverRevision: vault.serverRevision,
+    tombstone
+  });
+}
+
+function detectFileConflict(current: SyncFileRecord | undefined, incoming: SyncFileRecord): SyncConflict | undefined {
+  if (!current) {
+    return undefined;
+  }
+
+  if (current.manifest.contentHash === incoming.manifest.contentHash) {
+    return undefined;
+  }
+
+  if (!incoming.manifest.baseRevisionId || incoming.manifest.baseRevisionId !== current.manifest.revisionId) {
+    return {
+      path: incoming.manifest.path,
+      serverRevisionId: current.manifest.revisionId,
+      clientRevisionId: incoming.manifest.revisionId,
+      reason: "revision-mismatch"
+    };
+  }
+
+  return undefined;
+}
+
+function detectTombstoneConflict(current: SyncFileRecord | undefined, incoming: Tombstone): SyncConflict | undefined {
+  if (!current) {
+    return undefined;
+  }
+
+  if (!incoming.baseRevisionId || incoming.baseRevisionId !== current.manifest.revisionId) {
+    return {
+      path: incoming.path,
+      serverRevisionId: current.manifest.revisionId,
+      clientRevisionId: incoming.revisionId,
+      reason: "revision-mismatch"
+    };
+  }
+
+  return undefined;
+}
+
+export function createFileSyncStore(filePath: string): SyncStore {
+  const vaults = loadVaults(filePath);
+
+  function getVault(vaultId: string): VaultState {
+    let vault = vaults.get(vaultId);
+    if (!vault) {
+      vault = createVaultState();
+      vaults.set(vaultId, vault);
+    }
+
+    return vault;
+  }
+
+  function persist(): void {
+    persistVaults(filePath, vaults);
+  }
+
+  return {
+    registerDevice(request: RegisterDeviceRequest): RegisterDeviceResponse {
+      const vault = getVault(request.vaultId);
+      const deviceId = globalThis.crypto.randomUUID();
+      const token = globalThis.crypto.randomUUID();
+      const issuedAt = new Date().toISOString();
+
+      vault.devices.set(deviceId, {
+        deviceId,
+        deviceName: request.deviceName,
+        token,
+        issuedAt
+      });
+
+      persist();
+
+      return {
+        vaultId: request.vaultId,
+        deviceId,
+        token,
+        issuedAt
+      };
+    },
+
+    authenticate(vaultId: string, deviceId: string, token: string): boolean {
+      const vault = vaults.get(vaultId);
+      const device = vault?.devices.get(deviceId);
+      return Boolean(device && !device.revokedAt && device.token === token);
+    },
+
+    pull(vaultId: string, sinceServerRevision: number, limit?: number): SyncPullResponse {
+      const vault = getVault(vaultId);
+      const matchingChanges = vault.changes.filter((change) => change.serverRevision > sinceServerRevision);
+      const page = matchingChanges.slice(0, limit ?? matchingChanges.length);
+      const nextSinceServerRevision = page.at(-1)?.serverRevision ?? sinceServerRevision;
+
+      return {
+        serverRevision: vault.serverRevision,
+        changes: page,
+        hasMore: matchingChanges.length > page.length,
+        nextSinceServerRevision,
+        activeKeyId: vault.activeKeyId,
+        keyRotatedAt: vault.keyRotatedAt
+      };
+    },
+
+    push(request: SyncPushRequest): SyncPushResponse {
+      const vault = getVault(request.vaultId);
+      const acceptedFilePaths: string[] = [];
+      const acceptedTombstones: string[] = [];
+      const conflicts: SyncConflict[] = [];
+      let acceptedKeyId: string | undefined;
+
+      for (const file of request.files) {
+        const current = vault.files.get(file.manifest.path);
+        const conflict = detectFileConflict(current, file);
+        if (conflict) {
+          conflicts.push(conflict);
+          continue;
+        }
+
+        vault.files.set(file.manifest.path, file);
+        vault.tombstones.delete(file.manifest.path);
+        appendFileChange(vault, file);
+        acceptedFilePaths.push(file.manifest.path);
+        acceptedKeyId = file.envelope.keyId;
+      }
+
+      for (const tombstone of request.tombstones) {
+        const current = vault.files.get(tombstone.path);
+        const conflict = detectTombstoneConflict(current, tombstone);
+        if (conflict) {
+          conflicts.push(conflict);
+          continue;
+        }
+
+        vault.files.delete(tombstone.path);
+        vault.tombstones.set(tombstone.path, tombstone);
+        appendTombstoneChange(vault, tombstone);
+        acceptedTombstones.push(tombstone.path);
+      }
+
+      if (acceptedKeyId && acceptedKeyId !== vault.activeKeyId) {
+        vault.activeKeyId = acceptedKeyId;
+        vault.keyRotatedAt = new Date().toISOString();
+      }
+
+      if (acceptedFilePaths.length > 0 || acceptedTombstones.length > 0) {
+        persist();
+      }
+
+      return {
+        acceptedServerRevision: vault.serverRevision,
+        acceptedFilePaths,
+        acceptedTombstones,
+        conflicts,
+        activeKeyId: vault.activeKeyId,
+        keyRotatedAt: vault.keyRotatedAt
+      };
+    },
+
+    listDevices(vaultId: string): ListDevicesResponse {
+      const vault = getVault(vaultId);
+      const devices = [...vault.devices.values()].sort((left, right) => left.issuedAt.localeCompare(right.issuedAt));
+      return {
+        devices,
+        activeKeyId: vault.activeKeyId,
+        keyRotatedAt: vault.keyRotatedAt
+      };
+    },
+
+    revokeDevice(vaultId: string, targetDeviceId: string): RevokeDeviceResponse {
+      const vault = getVault(vaultId);
+      const device = vault.devices.get(targetDeviceId);
+      if (!device) {
+        throw new Error("Device not found.");
+      }
+
+      if (!device.revokedAt) {
+        device.revokedAt = new Date().toISOString();
+        vault.devices.set(targetDeviceId, device);
+        persist();
+      }
+
+      return {
+        targetDeviceId,
+        revokedAt: device.revokedAt
+      };
+    },
+
+    rotateVaultKey(vaultId: string, nextKeyId: string, previousKeyId?: string): RotateVaultKeyResponse {
+      const vault = getVault(vaultId);
+      if (previousKeyId && vault.activeKeyId && previousKeyId !== vault.activeKeyId) {
+        throw new Error("The provided previous key does not match the current server key.");
+      }
+
+      vault.activeKeyId = nextKeyId;
+      vault.keyRotatedAt = new Date().toISOString();
+      persist();
+
+      return {
+        activeKeyId: nextKeyId,
+        rotatedAt: vault.keyRotatedAt
+      };
+    }
+  };
+}

apps/sync-server/src/store/syncStore.ts

@@ -0,0 +1,20 @@
+import type {
+  ListDevicesResponse,
+  RegisterDeviceRequest,
+  RegisterDeviceResponse,
+  RevokeDeviceResponse,
+  RotateVaultKeyResponse,
+  SyncPullResponse,
+  SyncPushRequest,
+  SyncPushResponse
+} from "@obsidian-sync/sync-protocol";
+
+export interface SyncStore {
+  registerDevice(request: RegisterDeviceRequest): RegisterDeviceResponse;
+  authenticate(vaultId: string, deviceId: string, token: string): boolean;
+  pull(vaultId: string, sinceServerRevision: number, limit?: number): SyncPullResponse;
+  push(request: SyncPushRequest): SyncPushResponse;
+  listDevices(vaultId: string): ListDevicesResponse;
+  revokeDevice(vaultId: string, targetDeviceId: string): RevokeDeviceResponse;
+  rotateVaultKey(vaultId: string, nextKeyId: string, previousKeyId?: string): RotateVaultKeyResponse;
+}