Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -208,12 +208,12 @@ The provider manages the WebSocket lifecycle. When another user inserts a row, y

EdgePod helps you stay within Durable Object limits with lightweight, always-on guards:

| Guard | What it does |
| --------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Result limit** | Queries are capped at **1 000 rows**. If a query returns exactly 1 000 rows, a warning is logged to paginate with `.limit()` and `.offset()`. |
| **WHERE enforcement** | `UPDATE` and `DELETE` without a `.where()` clause are blocked. If you really mean to affect every row, chain `.withoutWhere()` to opt out per-query. |
| **Raw SQL guard** | Dangerous raw methods like `db.run()` and `db.get()` are blocked on the tracked database instance. Use `ctx.unsafeRawDb` explicitly if you need raw access, and call `ctx.invalidate()` manually. |
| **Bulk insert limit** | `insert().values()` arrays are capped at 1 000 rows to avoid oversized writes. |
| Guard | What it does |
| --------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Result limit** | Queries are capped at **1 000 rows**. If a query returns exactly 1 000 rows, a warning is logged to paginate with `.limit()` and `.offset()`. |
| **WHERE enforcement** | `UPDATE` and `DELETE` without a `.where()` clause are blocked. If you really mean to affect every row, chain `.withoutWhere()` to opt out per-query. |
| **Raw SQL guard** | Dangerous raw methods like `db.run()` and `db.get()` are blocked on the tracked database instance. Use `ctx.unsafeRawDb` explicitly if you need raw access — raw SQL is automatically tracked via SQL parsing. |
| **Bulk insert limit** | `insert().values()` arrays are capped at 1 000 rows to avoid oversized writes. |

These are not configuration options — they are designed to catch accidental misuse early, while giving you explicit escape hatches when you need them.

Expand Down
3 changes: 2 additions & 1 deletion packages/server/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,8 @@
"@logtape/logtape": "^2.0.5",
"drizzle-orm": "^0.45.2",
"jose": "^6.2.3",
"neverthrow": "^8.2.0"
"neverthrow": "^8.2.0",
"sqlite3-parser": "^0.7.1"
},
"devDependencies": {
"@types/better-sqlite3": "^7.6.13",
Expand Down
8 changes: 6 additions & 2 deletions packages/server/src/server/do.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -24,14 +24,18 @@ vi.mock("drizzle-orm/durable-sqlite/migrator", () => ({
migrate: vi.fn(),
}));

vi.mock("../tools/createTrackedDb", () => ({
createTrackedDb: vi.fn(),
vi.mock("../tools/createSafetyProxy", () => ({
createSafetyProxy: vi.fn(),
}));

vi.mock("../tools/buildCascadeGraph", () => ({
buildCascadeGraph: vi.fn(() => new Map()),
}));

vi.mock("../tools/buildPkMap", () => ({
buildPkMap: vi.fn(() => new Map()),
}));

vi.mock("./auth", () => ({
initJwtSigner: vi.fn(async () => ({ match: vi.fn() })),
getJwtSigner: vi.fn(() => null),
Expand Down
38 changes: 27 additions & 11 deletions packages/server/src/server/do.ts
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
import { DurableObject } from "cloudflare:workers";
import { drizzle } from "drizzle-orm/durable-sqlite";
import { migrate } from "drizzle-orm/durable-sqlite/migrator";
import { createTrackedDb } from "../tools/createTrackedDb";
import { createSafetyProxy } from "../tools/createSafetyProxy";
import { createTrackedRawDb } from "../tools/createTrackedRawDb";
import { buildCascadeGraph } from "../tools/buildCascadeGraph";
import { buildPkMap } from "../tools/buildPkMap";
import { initJwtSigner, getJwtSigner } from "./auth";
import { initLogger, createLogger } from "./logger";
import type { EdgePodSessionMap, EdgePodContext, RpcRequest, RpcMeta, JsonValue } from "../types";
Expand All @@ -19,6 +21,7 @@ export class BaseEdgePodEngine extends DurableObject {
private rawDb: ReturnType<typeof drizzle>;
private activeSessions: EdgePodSessionMap = new Map();
private cascadeGraph: Map<string, Set<string>> = new Map();
private pkMap: Map<string, string[]> = new Map();
protected userFunctions: Record<string, (...args: any[]) => Promise<JsonValue> | JsonValue> = {};
protected schema: Record<string, unknown> = {};
protected migrations: {
Expand All @@ -43,6 +46,7 @@ export class BaseEdgePodEngine extends DurableObject {
this.restoreActiveSessions();

this.cascadeGraph = buildCascadeGraph(this.schema);
this.pkMap = buildPkMap(this.schema);

await initLogger();
await initJwtSigner(this.env as any).match(
Expand Down Expand Up @@ -125,26 +129,31 @@ export class BaseEdgePodEngine extends DurableObject {
// Prepare the read/write trackers for this specific run
const tablesRead = new Set<string>();
const tablesWritten = new Set<string>();
const rowIds = new Map<string, Set<string>>();
const warnings: string[] = [];

// When reactive is false, pass an empty session map so no table subscriptions are registered
const sessionMap = reactive ? this.activeSessions : new Map();
const activeSessions = reactive ? this.activeSessions : new Map();

// Instantiate the Proxy
const dbProxy = createTrackedDb(
this.rawDb,
// Shared tracking context for this RPC invocation
const trackCtx = {
sessionId,
sessionMap,
activeSessions,
tablesRead,
tablesWritten,
this.cascadeGraph,
cascadeGraph: this.cascadeGraph,
warnings,
);
rowIds,
pkMap: this.pkMap,
};

// Instantiate the Proxy with SQL parser-based tracking
const dbProxy = createSafetyProxy(this.rawDb, trackCtx);

// Build the Context
const edgepodCtx: EdgePodContext<any, any, Record<string, any>> = {
db: dbProxy as any,
unsafeRawDb: this.rawDb,
unsafeRawDb: createTrackedRawDb(this.rawDb, trackCtx),
user,
env: this.env,
headers,
Expand All @@ -161,7 +170,6 @@ export class BaseEdgePodEngine extends DurableObject {
});
}
},
invalidate: (tables: string[]) => tables.forEach((t: string) => tablesWritten.add(t)),
set: (key: string, value: any) => variableStore.set(key, value),
get: (key: string) => variableStore.get(key) as any,
};
Expand All @@ -175,10 +183,18 @@ export class BaseEdgePodEngine extends DurableObject {
this.broadcastInvalidations(hashedTableNames);
}

const rowsMeta: Record<string, string[]> = {};
for (const [table, ids] of rowIds) {
rowsMeta[table] = [...ids];
}
return {
success: true,
data,
meta: { read: [...tablesRead], changed: [...tablesWritten] },
meta: {
read: [...tablesRead],
changed: [...tablesWritten],
...(Object.keys(rowsMeta).length > 0 ? { rows: rowsMeta } : {}),
},
warnings,
};
} catch (e) {
Expand Down
12 changes: 10 additions & 2 deletions packages/server/src/server/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ import pkg from "../../package.json" with { type: "json" };
import type { BaseEdgePodEngine } from "./do";
import type { RpcRequest } from "../types";
import { verifyJwt } from "./auth";
import { hashMetaTableNames } from "../tools/hashTableName";
import { hashMetaTableNames, hashTableName } from "../tools/hashTableName";
import { ResultAsync } from "neverthrow";

// EdgePod is origin-agnostic by design. Every request is authenticated via the
Expand Down Expand Up @@ -140,11 +140,19 @@ export const edgePodFetch = async (
});

if (result.success) {
const rowsMeta = result.meta.rows
? Object.fromEntries(
Object.entries(result.meta.rows).map(([t, ids]) => [hashTableName(t), ids]),
)
: undefined;
return Response.json(
{
success: true,
data: result.data,
_meta: { t: hashMetaTableNames(result.meta.read) },
_meta: {
t: hashMetaTableNames(result.meta.read),
...(rowsMeta ? { r: rowsMeta } : {}),
},
...(result.warnings.length > 0 ? { warnings: result.warnings } : {}),
},
{ headers: serverHeader },
Expand Down
13 changes: 13 additions & 0 deletions packages/server/src/tools/buildPkMap.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
import { getTableConfig } from "drizzle-orm/sqlite-core";

export function buildPkMap(schema: Record<string, unknown>): Map<string, string[]> {
const map = new Map<string, string[]>();
for (const key in schema) {
const table = schema[key];
if (!table || !(table as any)[Symbol.for("drizzle:Name")]) continue;
const config = getTableConfig(table as any);
const pkCols = config.columns.filter((c: any) => c.primary).map((c: any) => c.name);
Comment on lines +8 to +9
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 c.primary may not be the correct Drizzle column property

config.columns.filter((c) => c.primary) uses the .primary field to detect primary-key columns. In the Drizzle ORM SQLite adapter the canonical property exposed by getTableConfig is c.primaryKey (a boolean), not c.primary. If the property name is wrong, every pkCols array will be empty and pkMap will be populated but useless for future consumers. Worth verifying against the installed Drizzle version before pkMap is wired into invalidation logic.

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Proof that .primary is correct and .primaryKey is wrong:

  1. Constructor source (drizzle-orm/column.cjs:36):
    this.primary = config.primaryKey;
    The constructor maps config.primaryKey (config-time property) to this.primary (instance property).
  2. Runtime verification:
    const config = getTableConfig(users);
    for (const col of config.columns) {
    console.log(col.name, '| primary:', col.primary, '| primaryKey:', col.primaryKey);
    }
    Output:
    id | primary: true | primaryKey: undefined
    name | primary: false | primaryKey: undefined
  • col.primary → true/false (correct, works)
  • col.primaryKey → undefined always (wrong, broken)
    Greptile confused the config-time property name (primaryKey in the builder API: integer("id").primaryKey()) with the runtime instance property name (primary on the column object returned by getTableConfig). The code correctly uses the runtime property .primary.

map.set(config.name, pkCols);
}
return map;
}
40 changes: 0 additions & 40 deletions packages/server/src/tools/checkResultWarnings.test.ts

This file was deleted.

7 changes: 0 additions & 7 deletions packages/server/src/tools/checkResultWarnings.ts

This file was deleted.

139 changes: 139 additions & 0 deletions packages/server/src/tools/createBuilderProxy.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
import type { TrackContext } from "./createSafetyProxy";
import { trackExec, warnRowLimit } from "./tracking";

const STATE = Symbol("edgepod_builder_state");
const EXEC = ["then", "run", "all", "get", "values", "execute"];
const MAX_LIMIT = 1000;
const MAX_BULK_INSERT = 1000;

type BuilderConfig = {
type: "select" | "insert" | "update" | "delete";
tableName?: string;
};

export function createBuilderProxy(builder: any, ctx: TrackContext, config: BuilderConfig): any {
if (!builder[STATE]) {
builder[STATE] = { limitSet: false, whereSet: false, withoutWhereSet: false };
}

return new Proxy(builder, {
get(target: any, prop: string) {
const state: { limitSet: boolean; whereSet: boolean; withoutWhereSet: boolean } = target[
STATE
] || { limitSet: false, whereSet: false, withoutWhereSet: false };

// Safety: limit clamping for SELECT
if (prop === "limit" && config.type === "select")
return (n: number) => {
const clamped = Math.max(0, Math.min(n, MAX_LIMIT));
if (n > MAX_LIMIT) ctx.warnings.push(`Query limit of ${n} overridden to ${MAX_LIMIT}.`);
return wrap(target.limit(clamped), ctx, config, { ...state, limitSet: true });
};

// Safety: WHERE enforcement for UPDATE/DELETE
if (prop === "where" && (config.type === "update" || config.type === "delete"))
return (...args: unknown[]) =>
wrap(target.where(...args), ctx, config, { ...state, whereSet: true });
if (prop === "withoutWhere" && (config.type === "update" || config.type === "delete"))
return () => {
ctx.warnings.push(`[EdgePod] Unfiltered ${config.type} executed via .withoutWhere().`);
return wrap(target, ctx, config, { ...state, withoutWhereSet: true });
};
Comment thread
greptile-apps[bot] marked this conversation as resolved.

// Safety: bulk insert limit
if (prop === "values" && config.type === "insert")
return (...args: unknown[]) => {
const rows = args[0];
if (Array.isArray(rows) && rows.length > MAX_BULK_INSERT)
throw new Error(
`[EdgePod] Bulk insert blocked: ${rows.length} rows > ${MAX_BULK_INSERT}. Split into smaller batches.`,
);
return wrap(target.values(...args), ctx, config, state);
};

// Safety: prepare
if (prop === "prepare" && config.type !== "select")
return () => {
throw new Error(`[EdgePod] .prepare() is not supported for ${config.type}s.`);
};
if (prop === "prepare")
return (...args: unknown[]) => {
const b = state.limitSet ? target : target.limit(MAX_LIMIT);
return (b as any).prepare(...args);
};

// Execution: safety check → track → execute
if (EXEC.includes(prop))
return (...args: unknown[]) => {
if (
!state.whereSet &&
!state.withoutWhereSet &&
(config.type === "update" || config.type === "delete")
)
throw new Error(
`[EdgePod] ${config.type.toUpperCase()} without WHERE is blocked. If intentional, chain .withoutWhere().`,
);

let b = target;
if (config.type === "select" && !state.limitSet) b = target.limit(MAX_LIMIT);

trackExec(b, ctx, config.tableName, config.type);

if (prop === "then") {
const [resolve, reject] = args as [(v: unknown) => void, (e: unknown) => void];
return b.then((res: unknown) => {
warnRowLimit(res, ctx.warnings);
resolve(res);
}, reject);
}
Comment thread
greptile-apps[bot] marked this conversation as resolved.

const method = b[prop] as Function;
const result = method.apply(b, args);
warnRowLimit(result, ctx.warnings);
return result;
};

// Generic pass-through: wrap if result is a builder
return passThrough(target, prop, ctx, config, state);
},
});
}

function wrap(
result: any,
ctx: TrackContext,
config: BuilderConfig,
state: { limitSet: boolean; whereSet: boolean; withoutWhereSet: boolean },
): any {
if (isBuilder(result)) {
result[STATE] = state;
return createBuilderProxy(result, ctx, config);
}
return result;
}

function passThrough(
target: any,
prop: string,
ctx: TrackContext,
config: BuilderConfig,
state: { limitSet: boolean; whereSet: boolean; withoutWhereSet: boolean },
): any {
const raw = target[prop];
if (typeof raw === "function")
return (...args: unknown[]) => {
const result = raw.apply(target, args);
return isBuilder(result)
? createBuilderProxy(Object.assign(result, { [STATE]: state }), ctx, config)
: result;
};
if (isBuilder(raw)) {
raw[STATE] = state;
return createBuilderProxy(raw, ctx, config);
}
return raw;
}

function isBuilder(v: unknown): boolean {
return typeof v === "object" && v !== null && !Array.isArray(v);
}
Loading