Refactor quota and owner checks to be required callbacks

Changed `isOwnerAllowed` and `isOwnerWithinQuota` to required callbacks in relay and storage configs, removing their optional status.

Author: steida

apps/relay/src/index.ts

@@ -12,15 +12,12 @@ const relay = await createNodeJsRelay({
   port: 4000,
   enableLogging: false,
 
-  // Click to `isOwnerAllowed` to read the docs.
-  // isOwnerAllowed: (_ownerId) => true,
+  isOwnerAllowed: (_ownerId) => true,
 
-  // Click to `isOwnerWithinQuota` to read the docs.
-  // isOwnerWithinQuota: (ownerId, requiredBytes) => {
-  //   console.log(ownerId, requiredBytes);
-  //   // Check error via evolu.subscribeError
-  //   return true;
-  // },
+  isOwnerWithinQuota: (_ownerId, requiredBytes) => {
+    const maxBytes = 1024 * 1024; // 1MB
+    return requiredBytes <= maxBytes;
+  },
 });
 
 if (relay.ok) {

packages/common/src/Evolu/Relay.ts

@@ -35,11 +35,10 @@ export interface RelayConfig extends ConsoleConfig, StorageConfig {
   readonly name?: SimpleName;
 
   /**
-   * Optional callback to check if an {@link OwnerId} is allowed to access the
-   * relay. If this callback is not provided, all owners are allowed.
+   * Callback to check if an {@link OwnerId} is allowed to access the relay.
    *
-   * If provided, the callback receives the OwnerId and should return a
-   * {@link MaybeAsync} boolean: `true` to allow access, or `false` to deny.
+   * The callback receives the OwnerId and returns a {@link MaybeAsync} boolean:
+   * `true` to allow access, or `false` to deny.
    *
    * The callback can be synchronous (for SQLite or in-memory checks) or
    * asynchronous (for calling remote APIs).
@@ -76,7 +75,7 @@ export interface RelayConfig extends ConsoleConfig, StorageConfig {
    *   Promise.resolve(ownerId === "6jy_2F4RT5qqeLgJ14_dnQ"),
    * ```
    */
-  readonly isOwnerAllowed?: (ownerId: OwnerId) => MaybeAsync<boolean>;
+  readonly isOwnerAllowed: (ownerId: OwnerId) => MaybeAsync<boolean>;
 }
 
 /**
@@ -211,17 +210,15 @@ export const createRelaySqliteStorage =
               storedBytes + incomingBytes,
             );
 
-            if (config.isOwnerWithinQuota) {
-              const withinQuotaResult = config.isOwnerWithinQuota(
-                ownerId,
-                newStoredBytes,
-              );
-              const isWithinQuota = isAsync(withinQuotaResult)
-                ? await withinQuotaResult
-                : withinQuotaResult;
-              if (!isWithinQuota) {
-                return err({ type: "StorageQuotaError", ownerId });
-              }
+            const withinQuotaResult = config.isOwnerWithinQuota(
+              ownerId,
+              newStoredBytes,
+            );
+            const isWithinQuota = isAsync(withinQuotaResult)
+              ? await withinQuotaResult
+              : withinQuotaResult;
+            if (!isWithinQuota) {
+              return err({ type: "StorageQuotaError", ownerId });
             }
 
             return deps.sqlite.transaction(() => {

packages/common/src/Evolu/Storage.ts

@@ -29,13 +29,12 @@ import { orderTimestampBytes, Timestamp, TimestampBytes } from "./Timestamp.js";
 
 export interface StorageConfig {
   /**
-   * Optional callback to check if an {@link OwnerId} is within their quota for
-   * the requested write. If this callback is not provided, all writes are
-   * allowed regardless of size.
+   * Callback called before an attempt to write, to check if an {@link OwnerId}
+   * has sufficient quota for the write.
    *
-   * If provided, the callback receives the OwnerId and the number of bytes
-   * required for the write, and should return a {@link MaybeAsync} boolean:
-   * `true` to allow the write, or `false` to deny it due to quota limits.
+   * The callback receives the {@link OwnerId} and the number of bytes required
+   * for the write, and returns a {@link MaybeAsync} boolean: `true` to allow the
+   * write, or `false` to deny it due to quota limits.
    *
    * The callback can be synchronous (for SQLite or in-memory checks) or
    * asynchronous (for calling remote APIs).
@@ -58,7 +57,7 @@ export interface StorageConfig {
    * };
    * ```
    */
-  readonly isOwnerWithinQuota?: (
+  readonly isOwnerWithinQuota: (
     ownerId: OwnerId,
     requiredBytes: PositiveInt,
   ) => MaybeAsync<boolean>;

packages/common/src/Evolu/Sync.ts

@@ -9,7 +9,7 @@ import {
 } from "../Crypto.js";
 import { eqArrayNumber } from "../Eq.js";
 import { createTransferableError, TransferableError } from "../Error.js";
-import { constFalse } from "../Function.js";
+import { constFalse, constTrue } from "../Function.js";
 import { objectToEntries } from "../Object.js";
 import { RandomDep } from "../Random.js";
 import { createResources } from "../Resources.js";
@@ -451,6 +451,7 @@ const createClientStorage =
   }): Result<ClientStorage, SqliteError> => {
     const sqliteStorageBase = createBaseSqliteStorage(deps)({
       onStorageError: config.onError,
+      isOwnerWithinQuota: constTrue, // Clients don't have quota limits
     });
 
     // TODO: Mutex per OwnerId

packages/common/test/Evolu/Storage.test.ts

@@ -1,5 +1,6 @@
 import { sha256 } from "@noble/hashes/sha2.js";
 import { assert, expect, test } from "vitest";
+import { ownerIdToOwnerIdBytes } from "../../src/Evolu/Owner.js";
 import {
   BaseSqliteStorageDep,
   createBaseSqliteStorage,
@@ -14,20 +15,15 @@ import {
   createTimestamp,
   Millis,
   orderTimestampBytes,
+  TimestampBytes,
   timestampToTimestampBytes,
 } from "../../src/Evolu/Timestamp.js";
-import {
-  computeBalancedBuckets,
-  createRandom,
-  getOrThrow,
-  NonNegativeInt,
-  ok,
-  ownerIdToOwnerIdBytes,
-  PositiveInt,
-  sql,
-  SqliteDep,
-  TimestampBytes,
-} from "../../src/index.js";
+import { constTrue } from "../../src/Function.js";
+import { computeBalancedBuckets } from "../../src/Number.js";
+import { createRandom } from "../../src/Random.js";
+import { getOrThrow, ok } from "../../src/Result.js";
+import { sql, SqliteDep } from "../../src/Sqlite.js";
+import { NonNegativeInt, PositiveInt } from "../../src/Type.js";
 import { testCreateSqlite, testOwner2, testOwnerIdBytes } from "../_deps.js";
 import {
   testAnotherTimestampsAsc,
@@ -49,6 +45,7 @@ const createDeps = async (): Promise<SqliteDep & BaseSqliteStorageDep> => {
     onStorageError: (error) => {
       throw new Error(error.type);
     },
+    isOwnerWithinQuota: constTrue, // Allow all writes in tests
   });
   return { sqlite, storage };
 };

packages/common/test/_deps.ts

@@ -22,7 +22,7 @@ import {
   StorageConfig,
   StorageDep,
 } from "../src/Evolu/Storage.js";
-import { constFalse, constVoid } from "../src/Function.js";
+import { constFalse, constTrue, constVoid } from "../src/Function.js";
 import {
   createRandom,
   createRandomLibWithSeed,
@@ -319,6 +319,7 @@ export const testCreateRelayStorageAndSqliteDeps = async (
     onStorageError: (error) => {
       throw new Error(error.type);
     },
+    isOwnerWithinQuota: constTrue, // Allow all writes in tests by default
     ...config,
   });
 

packages/nodejs/src/Evolu/Relay.ts

@@ -96,7 +96,7 @@ const createNodeJsRelayWithDeps =
 
     const storage = createRelaySqliteStorage(depsWithSqlite)({
       onStorageError: log.storageError,
-      ...(isOwnerWithinQuota && { isOwnerWithinQuota }),
+      isOwnerWithinQuota,
     });
 
     const server = createServer();
@@ -110,18 +110,6 @@ const createNodeJsRelayWithDeps =
     server.on("upgrade", (request, socket, head) => {
       socket.on("error", log.upgradeSocketError);
 
-      const completeUpgrade = () => {
-        socket.removeListener("error", log.upgradeSocketError);
-        wss.handleUpgrade(request, socket, head, (ws) => {
-          wss.emit("connection", ws, request);
-        });
-      };
-
-      if (!isOwnerAllowed) {
-        completeUpgrade();
-        return;
-      }
-
       const ownerId = parseOwnerIdFromOwnerWebSocketTransportUrl(
         request.url ?? "",
       );
@@ -142,7 +130,10 @@ const createNodeJsRelayWithDeps =
           socket.destroy();
           return;
         }
-        completeUpgrade();
+        socket.removeListener("error", log.upgradeSocketError);
+        wss.handleUpgrade(request, socket, head, (ws) => {
+          wss.emit("connection", ws, request);
+        });
       })();
     });