diff --git a/.changeset/tall-seals-sink.md b/.changeset/tall-seals-sink.md
new file mode 100644
index 0000000..eb94d47
--- /dev/null
+++ b/.changeset/tall-seals-sink.md
@@ -0,0 +1,5 @@
+---
+"@atmo-dev/contrail": patch
+---
+
+add postgres adapter and example
diff --git a/src/adapters/postgres.ts b/src/adapters/postgres.ts
index 96a8aab..ae75c9f 100644
--- a/src/adapters/postgres.ts
+++ b/src/adapters/postgres.ts
@@ -8,21 +8,37 @@ interface PgStatement extends Statement {
   _runOn(client: pg.PoolClient): Promise<any>;
 }
 
+/** Column names known to be BIGINT — PostgreSQL returns these as strings */
+const BIGINT_COLUMNS = new Set(["time_us", "indexed_at", "resolved_at"]);
+
 function normalizeRow(row: any): any {
-  if (row && typeof row.record === "object" && row.record !== null) {
+  if (!row) return row;
+  if (typeof row.record === "object" && row.record !== null) {
     row.record = JSON.stringify(row.record);
   }
-  // PostgreSQL returns BIGINT as string — coerce numeric fields back to number
-  if (row && typeof row.time_us === "string") row.time_us = Number(row.time_us);
-  if (row && typeof row.indexed_at === "string") row.indexed_at = Number(row.indexed_at);
-  if (row && typeof row.resolved_at === "string") row.resolved_at = Number(row.resolved_at);
+  for (const col of BIGINT_COLUMNS) {
+    if (typeof row[col] === "string") row[col] = Number(row[col]);
+  }
   return row;
 }
 
 export function createPostgresDatabase(pool: pg.Pool): Database {
   function rewritePlaceholders(sql: string): string {
     let idx = 0;
-    return sql.replace(/\?/g, () => `$${++idx}`);
+    let inString = false;
+    let result = "";
+    for (let i = 0; i < sql.length; i++) {
+      const ch = sql[i];
+      if (ch === "'" && sql[i - 1] !== "\\") {
+        inString = !inString;
+        result += ch;
+      } else if (ch === "?" && !inString) {
+        result += `$${++idx}`;
+      } else {
+        result += ch;
+      }
+    }
+    return result;
   }
 
   function wrapStatement(sql: string, boundValues: any[] = []): PgStatement {
diff --git a/src/core/dialect.ts b/src/core/dialect.ts
index 8c50dbb..e3bb790 100644
--- a/src/core/dialect.ts
+++ b/src/core/dialect.ts
@@ -3,6 +3,14 @@ export function getDialect(db: { dialect?: SqlDialect }): SqlDialect {
   return db.dialect ?? sqliteDialect;
 }
 
+const SAFE_FIELD = /^[a-zA-Z0-9_.]+$/;
+
+function assertSafeField(field: string): void {
+  if (!SAFE_FIELD.test(field)) {
+    throw new Error(`Invalid field name: ${field}`);
+  }
+}
+
 export interface SqlDialect {
   /** json_extract(col, '$.field') or col->>'field' */
   jsonExtract(column: string, field: string): string;
@@ -19,9 +27,6 @@ export interface SqlDialect {
   /** FTS strategy: 'virtual-table' (SQLite FTS5) or 'generated-column' (PG tsvector) */
   readonly ftsStrategy: "virtual-table" | "generated-column";
 
-  /** Placeholder for parameterized queries: ? (SQLite) or $N (PostgreSQL) */
-  placeholder(index: number): string;
-
   /** INTEGER type name — same on both, but PostgreSQL may want BIGINT for time_us */
   readonly integerType: string;
 
@@ -34,6 +39,7 @@ export interface SqlDialect {
 
 export const sqliteDialect: SqlDialect = {
   jsonExtract(column: string, field: string): string {
+    assertSafeField(field);
     return `json_extract(${column}, '$.${field}')`;
   },
 
@@ -43,11 +49,6 @@ export const sqliteDialect: SqlDialect = {
 
   recordColumnType: "TEXT",
   ftsStrategy: "virtual-table",
-
-  placeholder(_index: number): string {
-    return "?";
-  },
-
   integerType: "INTEGER",
   bigintType: "INTEGER",
 
@@ -58,6 +59,7 @@ export const sqliteDialect: SqlDialect = {
 
 export const postgresDialect: SqlDialect = {
   jsonExtract(column: string, field: string): string {
+    assertSafeField(field);
     const parts = field.split(".");
     if (parts.length === 1) {
       return `${column}->>'${parts[0]}'`;
@@ -73,11 +75,6 @@ export const postgresDialect: SqlDialect = {
 
   recordColumnType: "JSONB",
   ftsStrategy: "generated-column",
-
-  placeholder(index: number): string {
-    return `$${index}`;
-  },
-
   integerType: "INTEGER",
   bigintType: "BIGINT",
 
diff --git a/src/core/persistent.ts b/src/core/persistent.ts
index c541b51..e0945f1 100644
--- a/src/core/persistent.ts
+++ b/src/core/persistent.ts
@@ -175,11 +175,18 @@ async function streamAndFlush(
       // Race the next event against abort signal
       let result: IteratorResult<any>;
       if (signal) {
-        const abortPromise = new Promise<IteratorResult<any>>((resolve) => {
-          const handler = () => resolve({ value: undefined, done: true });
-          signal.addEventListener("abort", handler, { once: true });
-        });
-        result = await Promise.race([iterator.next(), abortPromise]);
+        const nextPromise = iterator.next();
+        if (signal.aborted) {
+          result = { value: undefined, done: true };
+        } else {
+          let abortHandler: () => void;
+          const abortPromise = new Promise<IteratorResult<any>>((resolve) => {
+            abortHandler = () => resolve({ value: undefined, done: true });
+            signal.addEventListener("abort", abortHandler, { once: true });
+          });
+          result = await Promise.race([nextPromise, abortPromise]);
+          signal.removeEventListener("abort", abortHandler!);
+        }
       } else {
         result = await iterator.next();
       }
diff --git a/tests/dialect.test.ts b/tests/dialect.test.ts
index ccb5dd1..b8343b2 100644
--- a/tests/dialect.test.ts
+++ b/tests/dialect.test.ts
@@ -36,10 +36,6 @@ describe("sqliteDialect", () => {
     expect(sqliteDialect.ftsStrategy).toBe("virtual-table");
   });
 
-  it("uses ? placeholders", () => {
-    expect(sqliteDialect.placeholder(1)).toBe("?");
-    expect(sqliteDialect.placeholder(5)).toBe("?");
-  });
 });
 
 describe("Database.dialect", () => {
@@ -90,10 +86,6 @@ describe("postgresDialect", () => {
     expect(postgresDialect.ftsStrategy).toBe("generated-column");
   });
 
-  it("uses $N placeholders", () => {
-    expect(postgresDialect.placeholder(1)).toBe("$1");
-    expect(postgresDialect.placeholder(3)).toBe("$3");
-  });
 });
 
 describe("indexExpression", () => {