ESAV Live

.gitignore

···
 deno.lock
 reindex.sh
+scripts/reindex.ts

src/firehose.ts

···
 import { AppConfig } from "./config.ts";
-import { indexDocument, deleteDocument } from "./indexer.ts";
+import { indexDocument, deleteDocument, type OnEventCallback } from "./indexer.ts";
 
-export async function startFirehose(config: AppConfig) {
-  const ws = new WebSocket(config.jetstream_url);
+interface FirehoseOptions {
+  config: AppConfig;
+  onEvent: OnEventCallback;
+}
+
+export function startFirehose({ config, onEvent }: FirehoseOptions) {
+  let lastCursor: number | null = null;
 
-  ws.onopen = () => {
-    console.log("Connected to Jetstream");
-  };
+  const connect = () => {
+    const url = new URL(config.jetstream_url);
+    if (lastCursor !== null) {
+      url.searchParams.set("cursor", lastCursor.toString());
+    }
 
-  ws.onmessage = async (msg) => {
-    const data = msg.data instanceof Blob ? await msg.data.text() : msg.data;
-    const evt = JSON.parse(data);
+    const ws = new WebSocket(url.toString());
+
+    ws.onopen = () => {
+      console.log("Connected to Jetstream", lastCursor !== null ? `(resuming from cursor ${lastCursor})` : "");
+    };
+
+    ws.onmessage = async (msg) => {
+      const data = msg.data instanceof Blob ? await msg.data.text() : msg.data;
+      const evt = JSON.parse(data);
+
+      if (!evt?.commit) return;
+
+      lastCursor = evt.time_us;
+
+      const opType = evt.commit.operation;
+      const cid = evt.commit.cid;
+      const collection = evt.commit.collection;
+      const rkey = evt.commit.rkey;
+      const aturi = `at://${evt.did}/${collection}/${rkey}`;
+      const record = evt.commit.record;
+      const indexedAt = new Date(evt.time_us / 1000).toISOString();
 
-    if (!evt?.commit) return;
+      if (!config.record_types.includes(collection)) return;
 
-    const opType = evt.commit.operation;
-    const cid = evt.commit.cid;
-    const collection = evt.commit.collection;
-    const rkey = evt.commit.rkey;
-    const aturi = `at://${evt.did}/${collection}/${rkey}`;
-    const record = evt.commit.record;
-    const indexedAt = new Date(evt.time_us / 1000).toISOString();
+      if (opType === "create" || opType === "update") {
+        await indexDocument(config, onEvent, aturi, record, cid, indexedAt);
+      } else if (opType === "delete") {
+        await deleteDocument(config, onEvent, aturi);
+      }
+    };
 
-    if (!config.record_types.includes(collection)) return;
+    ws.onerror = (e) => {
+      console.error("Jetstream WebSocket error:", e);
+    };
 
-    if (opType === "create" || opType === "update") {
-      await indexDocument(config, aturi, record, cid, indexedAt);
-    } else if (opType === "delete") {
-      await deleteDocument(config, aturi);
-    }
+    ws.onclose = (e) => {
+      console.warn("Jetstream disconnected. Attempting to reconnect in 1 second...", e.reason);
+      setTimeout(connect, 1000);
+    };
   };
 
-  ws.onerror = (e) => {
-    console.error("Jetstream error:", e);
-  };
+  connect();
 }

src/indexer.ts

···
   };
 }
 
+function flattenValues(input: unknown): unknown[] {
+  const result: unknown[] = [];
+
+  function recurse(value: unknown) {
+    if (value === null || typeof value !== "object") {
+      result.push(value);
+    } else if (Array.isArray(value)) {
+      for (const item of value) recurse(item);
+    } else if (value) {
+      for (const key in value) {
+        recurse((value as Record<string, unknown>)[key]);
+      }
+    }
+  }
+
+  recurse(input);
+  return result;
+}
+
 export async function ensureIndexMapping(config: AppConfig) {
   const properties: Record<string, any> = {};
 
···
 ): Record<string, unknown> {
   const result: Record<string, unknown> = {};
 
-  for (const [jsonPath, { id }] of Object.entries(indexSpec)) {
-    const pathSegments = jsonPath.split(".");
+  for (const [rawPath, { id }] of Object.entries(indexSpec)) {
+    const [path, modifier] = rawPath.split("#");
+    const pathSegments = path.split(".");
+
     let value = record;
     for (const segment of pathSegments) {
       if (typeof value !== "object" || value === null) {
···
       value = value[segment];
     }
 
-    if (value !== undefined) {
+    if (value === undefined) continue;
+
+    if (modifier === "flatten") {
+      const flattened = flattenValues(value);
+      if (flattened.length > 0) {
+        result[id] = flattened;
+      }
+    } else {
       result[id] = value;
     }
   }
···
   return result;
 }
 
+export type IndexerEvent = 
+  | { type: 'index'; uri: string; data: Record<string, unknown> }
+  | { type: 'delete'; uri: string };
+
+export type OnEventCallback = (event: IndexerEvent) => Promise<void> | void;
+
 export async function indexDocument(
   config: AppConfig,
+  onEvent: OnEventCallback,
   uri: string,
-  record: any,
+  record: Record<string, unknown>,
   cid: string,
   indexedAt: string
 ) {
···
   const recordType = collection;
 
   const indexSpec = config.index_fields?.[recordType];
-  if (!indexSpec) return; // skip if record type not configured
+  //if (!indexSpec) return;
 
   const indexedFields = extractIndexedFields(record, indexSpec);
 
···
       body: JSON.stringify(body),
     }
   );
+  await onEvent({
+    type: 'index',
+    uri,
+    data: body
+  });
 
   if (!res.ok) {
     console.error("Indexing failed:", await res.text());
   }
 }
 
-export async function deleteDocument(config: AppConfig, uri: string) {
+export async function deleteDocument(
+  config: AppConfig, 
+  onEvent: OnEventCallback, 
+  uri: string
+) {
   const res = await fetch(
     `${config.es_url}/${config.index_name}/_doc/${encodeURIComponent(uri)}`,
     {
       method: "DELETE",
     }
   );
+  await onEvent({
+    type: 'delete',
+    uri
+  });
 
   if (!res.ok && res.status !== 404) {
     console.error("Delete failed:", await res.text());

src/live-utils.ts

···
+type QueryClause = {
+  match?: Record<string, unknown>;
+  term?: Record<string, unknown>;
+  bool?: {
+    must?: QueryClause[];
+    must_not?: QueryClause[];
+    should?: QueryClause[];
+    filter?: QueryClause[];
+  };
+  range?: Record<string, Record<string, unknown>>;
+  nested?: {
+    path: string;
+    query: QueryClause;
+  };
+  exists?: {
+    field: string;
+  };
+};
+
+// function getField(obj: Record<string, any>, path: string): any {
+//     return path.split('.').reduce((acc, part) => acc && acc[part], obj);
+// }
+
+export function getSafeField(obj: Record<string, any>, path: string): any {
+  if (obj && typeof obj === 'object' && Object.prototype.hasOwnProperty.call(obj, path)) {
+    return obj[path];
+  }
+  return path.split('.').reduce((acc, part) => acc?.[part], obj);
+}
+
+export function matchDocAgainstQuery(doc: Record<string, unknown>, query: QueryClause): boolean {
+  console.log(JSON.stringify(doc, null, 2));
+  if (!query || Object.keys(query).length === 0) {
+    console.log("✅ Accepted: No query provided, accepting all docs.");
+    return true;
+  }
+
+  for (const key in query) {
+    const clauseValue = (query as any)[key];
+
+    switch (key) {
+      case 'term': {
+        const [field, value] = Object.entries(clauseValue as object)[0];
+        const docValue = getSafeField(doc, field);
+        if (docValue !== value) {
+          console.log(`❌ Rejected: term mismatch on field '${field}', expected '${value}', got '${docValue}'`);
+          return false;
+        }
+        console.log(`✅ Passed: term match on field '${field}' with value '${value}'`);
+        break;
+      }
+
+      case 'match': {
+        const [field, value] = Object.entries(clauseValue as object)[0];
+        const fieldValue = getSafeField(doc, field);
+        if (typeof fieldValue !== 'string') {
+          console.log(`❌ Rejected: match expected string field '${field}', got non-string:`, fieldValue);
+          return false;
+        }
+        if (!fieldValue.includes(String(value))) {
+          console.log(`❌ Rejected: match failed on field '${field}', expected to include '${value}', got '${fieldValue}'`);
+          return false;
+        }
+        console.log(`✅ Passed: match succeeded on field '${field}' includes '${value}'`);
+        break;
+      }
+
+      case 'bool': {
+        const boolQuery = clauseValue as QueryClause['bool'];
+
+        if (boolQuery?.must?.some(q => !matchDocAgainstQuery(doc, q))) {
+          console.log("❌ Rejected: bool.must clause failed.");
+          return false;
+        }
+        if (boolQuery?.filter?.some(q => !matchDocAgainstQuery(doc, q))) {
+          console.log("❌ Rejected: bool.filter clause failed.");
+          return false;
+        }
+        if (boolQuery?.must_not?.some(q => matchDocAgainstQuery(doc, q))) {
+          console.log("❌ Rejected: bool.must_not clause matched (should not have).");
+          return false;
+        }
+        if (boolQuery?.should?.length && !boolQuery.should.some(q => matchDocAgainstQuery(doc, q))) {
+          console.log("❌ Rejected: bool.should clause did not match any.");
+          return false;
+        }
+        console.log("✅ Passed: bool clause matched.");
+        break;
+      }
+
+      case 'range': {
+        const [field, range] = Object.entries(clauseValue as object)[0];
+        const value = getSafeField(doc, field);
+        if (value === undefined) {
+          console.log(`❌ Rejected: range field '${field}' is undefined in doc.`);
+          return false;
+        }
+
+        for (const [op, opValue] of Object.entries(range as object)) {
+          const numValue = Number(value);
+          const numOpValue = Number(opValue);
+          const isNumeric = !isNaN(numValue) && !isNaN(numOpValue);
+          let passed = true;
+
+          switch (op) {
+            case 'gt':
+              passed = isNumeric ? numValue > numOpValue : String(value) > String(opValue);
+              break;
+            case 'gte':
+              passed = isNumeric ? numValue >= numOpValue : String(value) >= String(opValue);
+              break;
+            case 'lt':
+              passed = isNumeric ? numValue < numOpValue : String(value) < String(opValue);
+              break;
+            case 'lte':
+              passed = isNumeric ? numValue <= numOpValue : String(value) <= String(opValue);
+              break;
+          }
+
+          if (!passed) {
+            console.log(`❌ Rejected: range.${op} failed on field '${field}', value '${value}' vs '${opValue}'`);
+            return false;
+          }
+        }
+
+        console.log(`✅ Passed: range match on field '${field}'`);
+        break;
+      }
+
+      case 'exists': {
+        const field = (clauseValue as { field: string }).field;
+        if (getSafeField(doc, field) === undefined) {
+          console.log(`❌ Rejected: exists failed, field '${field}' not found`);
+          return false;
+        }
+        console.log(`✅ Passed: exists matched, field '${field}' exists`);
+        break;
+      }
+
+      default:
+        console.log(`ℹ️ Ignored unknown query clause: '${key}'`);
+        break;
+    }
+  }
+
+  console.log("✅ Accepted: Document matched all clauses.");
+  return true;
+}
+
+const ALLOWED_QUERY_CLAUSES = new Set([
+  'term',
+  'terms',
+  'bool',
+  'range',
+  'exists',
+]);
+
+const FORBIDDEN_QUERY_CLAUSES = new Set([
+  'match',
+  'multi_match',
+  'match_phrase',
+  'query_string',
+  'simple_query_string',
+  'fuzzy',
+  'script',
+  'function_score',
+  'more_like_this',
+  'percolate',
+]);
+
+function validateClause(clause: Record<string, unknown>) {
+  for (const key in clause) {
+    if (FORBIDDEN_QUERY_CLAUSES.has(key)) {
+      throw new Error(`Query clause '${key}' is not allowed for live sync due to non-deterministic behavior. Please use deterministic clauses like 'term' or 'range'.`);
+    }
+
+    if (!ALLOWED_QUERY_CLAUSES.has(key)) {
+      throw new Error(`Query clause '${key}' is not supported for live sync.`);
+    }
+
+    const value = clause[key];
+    if (key === 'bool' && typeof value === 'object' && value !== null) {
+      const boolClauses = value as Record<string, unknown[]>;
+      for (const boolType of ['must', 'filter', 'should', 'must_not']) {
+        if (Array.isArray(boolClauses[boolType])) {
+          for (const subClause of boolClauses[boolType]) {
+            if (typeof subClause === 'object' && subClause !== null) {
+              validateClause(subClause as Record<string, unknown>);
+            }
+          }
+        }
+      }
+    }
+  }
+}
+
+export function validateLiveQuery(esQuery: Record<string, unknown>): void {
+  // a query must have an explicit 'sort' clause
+  // which prevents non-deterministic sorting (bad)
+  const sortClause = esQuery.sort as any[];
+  if (!Array.isArray(sortClause) || sortClause.length === 0) {
+    throw new Error("Live queries must include an explicit 'sort' clause for deterministic ordering (e.g., sort: [{ '$metadata.indexedAt': 'desc' }]).");
+  }
+
+  // and also of the supported deterministic clauses (no text searches)
+  const queryPart = esQuery.query as Record<string, unknown>;
+  if (queryPart) {
+    validateClause(queryPart);
+  }
+}

src/main.ts

···
 import { readConfig } from "./config.ts";
 import { startFirehose } from "./firehose.ts";
-import { ensureIndexMapping } from "./indexer.ts";
+import { ensureIndexMapping, type IndexerEvent } from "./indexer.ts";
 import { setupXRPCServer } from "./xrpc.ts";
+import { processEventForSync } from "./sync.ts";
 
-const config = await readConfig("./config.json");
+async function main() {
+  const config = await readConfig("./config.json");
 
-// prepare indexes
-ensureIndexMapping(config);
+  // prepare indexes
+  await ensureIndexMapping(config);
 
-startFirehose(config);
+  setupXRPCServer(config);
 
-setupXRPCServer(config);
+  startFirehose({
+    config,
+    onEvent: (event: IndexerEvent) => {
+      // ESAV Live !!!
+      return processEventForSync(event);
+    }
+  });
+
+  console.log("Server started and listening for events");
+}
+
+main().catch(err => {
+  console.error("Fatal error in main:", err);
+  Deno.exit(1);
+});

src/sync.ts

···
+import { subscriptions, computeCid, type Subscription, type SubscriptionResult } from "./xrpc.ts";
+import { type IndexerEvent } from "./indexer.ts";
+import { getSafeField, matchDocAgainstQuery } from "./live-utils.ts";
+
+interface ClientDelta {
+  documents: Record<string, { cid: string; doc: Record<string, unknown> }>;
+  queries: Record<string, { ecid: string; result: string[] }>;
+}
+
+function extractSortValues(doc: Record<string, any>, sortClause: any[]): unknown[] {
+  const values: unknown[] = [];
+  for (const sortField of sortClause) {
+    const key = Object.keys(sortField)[0];
+    const value = getSafeField(doc, key);
+    values.push(value);
+  }
+  return values;
+}
+
+function compareItems(aSorts: unknown[], bSorts: unknown[], sortClause: any[]): number {
+  for (let i = 0; i < sortClause.length; i++) {
+    const direction = Object.values(sortClause[i])[0];
+    const valA = aSorts[i];
+    const valB = bSorts[i];
+
+    if (valA === valB) continue;
+
+    if (valA === undefined || valA === null) return 1;
+    if (valB === undefined || valB === null) return -1;
+    
+    let comparison = 0;
+    if (valA < valB) {
+      comparison = -1;
+    } else if (valA > valB) {
+      comparison = 1;
+    }
+    
+    if (comparison !== 0) {
+      return direction === 'desc' ? -comparison : comparison;
+    }
+  }
+  return 0;
+}
+
+function findInsertIndex(sortedArray: Subscription['result'], newItem: SubscriptionResult, sortClause: any[]): number {
+    let low = 0;
+    let high = sortedArray.length;
+
+    while (low < high) {
+        const mid = Math.floor((low + high) / 2);
+        const comparison = compareItems(newItem.sortValues, sortedArray[mid].sortValues, sortClause);
+        if (comparison < 0) {
+            high = mid;
+        } else {
+            low = mid + 1;
+        }
+    }
+    return low;
+}
+
+export async function processEventForSync(event: IndexerEvent) {
+  const requiredDocuments = new Set<string>();
+  
+  const changedQueries: {
+      queryId: string;
+      newEcid: string;
+      newRichResult: SubscriptionResult[]; 
+      newResultUris: string[];
+      sub: Subscription;
+  }[] = [];
+
+  for (const [queryId, sub] of subscriptions.entries()) {
+    let newRichResult: SubscriptionResult[] | undefined = undefined;
+    const sortClause = (sub.esQuery as any).sort;
+
+    if (event.type === 'index') {
+      const queryClause = (sub.esQuery as any).query ?? {};
+      const doesMatch = matchDocAgainstQuery(event.data, queryClause);
+      const currentIndex = sub.result.findIndex(item => item.uri === event.uri);
+      const wasInResult = currentIndex !== -1;
+
+      if (doesMatch && !wasInResult) {
+        const newItem = {
+          uri: event.uri,
+          sortValues: extractSortValues(event.data, sortClause),
+        };
+        const insertIndex = findInsertIndex(sub.result, newItem, sortClause);
+        
+        newRichResult = [...sub.result];
+        newRichResult.splice(insertIndex, 0, newItem);
+        requiredDocuments.add(event.uri);
+
+      } else if (!doesMatch && wasInResult) {
+        newRichResult = sub.result.filter(item => item.uri !== event.uri);
+
+      } else if (doesMatch && wasInResult) {
+        const newItem = {
+          uri: event.uri,
+          sortValues: extractSortValues(event.data, sortClause),
+        };
+        if (compareItems(newItem.sortValues, sub.result[currentIndex].sortValues, sortClause) !== 0) {
+            const tempResult = sub.result.filter(item => item.uri !== event.uri);
+            const insertIndex = findInsertIndex(tempResult, newItem, sortClause);
+            
+            newRichResult = [...tempResult];
+            newRichResult.splice(insertIndex, 0, newItem);
+        }
+        requiredDocuments.add(event.uri);
+      }
+    } else if (event.type === 'delete') {
+      if (sub.result.some(item => item.uri === event.uri)) {
+        newRichResult = sub.result.filter(item => item.uri !== event.uri);
+      }
+    }
+
+    if (newRichResult) {
+      const newResultUris = newRichResult.map(item => item.uri);
+      const newEcid = await computeCid(newResultUris);
+      
+      if (newEcid !== sub.ecid) {
+          changedQueries.push({
+              queryId,
+              newEcid,
+              newRichResult,
+              newResultUris,
+              sub
+          });
+      }
+    }
+  }
+
+  for (const { queryId, newEcid, newRichResult } of changedQueries) {
+    const sub = subscriptions.get(queryId);
+    if (sub) {
+        sub.ecid = newEcid;
+        sub.result = newRichResult;
+    }
+  }
+
+  const documentsPayload: Record<string, { cid: string; doc: Record<string, unknown> }> = {};
+  if (event.type === 'index') {
+    for (const uri of requiredDocuments) {
+        documentsPayload[uri] = {
+            cid: event.data["$metadata.cid"] as string,
+            doc: event.data as Record<string, unknown>,
+        };
+    }
+  }
+
+  if (changedQueries.length > 0 || Object.keys(documentsPayload).length > 0) {
+    const affectedClients = new Set<WebSocket>();
+    changedQueries.forEach(cq => cq.sub.clients.forEach(c => affectedClients.add(c)));
+    
+    if (Object.keys(documentsPayload).length > 0 && event.type === 'index') {
+        for (const [, sub] of subscriptions.entries()) {
+            if (sub.result.some(item => item.uri === event.uri)) {
+                sub.clients.forEach(c => affectedClients.add(c));
+            }
+        }
+    }
+
+    for (const client of affectedClients) {
+        const delta: ClientDelta = {
+            documents: documentsPayload,
+            queries: {},
+        };
+
+        for (const { queryId, newEcid, newResultUris, sub } of changedQueries) {
+            if (sub.clients.has(client)) {
+                delta.queries[queryId] = { ecid: newEcid, result: newResultUris };
+            }
+        }
+        
+        if (Object.keys(delta.queries).length > 0 || Object.keys(documentsPayload).length > 0) {
+            if (client.readyState === WebSocket.OPEN) {
+                try {
+                    client.send(JSON.stringify({ type: 'query-delta', ...delta }));
+                } catch (err) {
+                    console.error("Failed to send delta to client:", err);
+                }
+            }
+        }
+    }
+  }
+}

src/xrpc.ts

···
-import { AppConfig } from "../src/config.ts";
+import { AppConfig } from "./config.ts";
 import { serve } from "https://deno.land/std/http/server.ts";
+import { CID } from "npm:multiformats/cid";
+import { sha256 } from "npm:multiformats/hashes/sha2";
+import { DidResolver, HandleResolver } from "npm:@atproto/identity";
+import * as dagCbor from "npm:@ipld/dag-cbor";
+import { validateLiveQuery } from "./live-utils.ts";
+
+const ZOMBIE_LIFETIME_MS = 5 * 60 * 1000;
+
+declare global {
+  interface WebSocket {
+    isAlive: boolean;
+    ping(): void;
+    terminate?(): void;
+  }
+}
+
+export interface SubscriptionResult {
+  uri: string;
+  sortValues: unknown[];
+}
+
+export interface Subscription {
+  esQuery: Record<string, unknown>;
+  clients: Set<WebSocket>;
+  ecid: string;
+  result: SubscriptionResult[];
+  zombieTimer?: number;
+}
+
+export const subscriptions = new Map<string, Subscription>();
+
+export async function computeCid(data: unknown): Promise<string> {
+  const encodedBytes = dagCbor.encode(data);
+  const hash = await sha256.digest(encodedBytes);
+  return CID.create(1, dagCbor.code, hash).toString();
+}
+
+function corsHeaders(): HeadersInit {
+  return {
+    "Access-Control-Allow-Origin": "*",
+    "Access-Control-Allow-Methods": "GET,POST,OPTIONS",
+    "Access-Control-Allow-Headers": "Content-Type",
+  };
+}
 
 export function setupXRPCServer(config: AppConfig) {
-  serve(async (req) => {
-    const url = new URL(req.url);
+  serve(
+    async (req) => {
+      const { pathname, searchParams } = new URL(req.url);
 
-    if (req.method === "OPTIONS") {
-      return new Response(null, {
-        status: 204,
-        headers: corsHeaders(),
-      });
-    }
+      if (req.method === "OPTIONS") {
+        return new Response(null, { status: 204, headers: corsHeaders() });
+      }
 
-    if (url.pathname === "/xrpc/com.example.prototypeESQuery") {
-      let esQuery: any;
+      if (
+        pathname === "/xrpc/party.whey.esav.esSync" &&
+        req.headers.get("upgrade") === "websocket"
+      ) {
+        const { socket, response } = Deno.upgradeWebSocket(req);
+        handleWebSocket(socket, config);
+        return response;
+      }
 
-      if (req.method === "POST") {
+      if (pathname === "/xrpc/party.whey.esav.resolveIdentity") {
         try {
-          esQuery = await req.json();
-        } catch {
-          return new Response("Invalid JSON body", {
-            status: 400,
+          const handle = searchParams.get("handle");
+          const did = searchParams.get("did");
+          const clientCid = searchParams.get("cid");
+          const includePfp = searchParams.get("includePfp");
+
+          if (!handle && !did) {
+            return new Response("handle or did parameter is required", {
+              status: 400,
+              headers: corsHeaders(),
+            });
+          }
+
+          const identity = await resolveWithHandleOrDid({
+            handle: handle ?? undefined,
+            did: did ?? undefined,
+          });
+
+          const finalResponse: Record<string, unknown> = {
+            ...identity,
+          };
+
+          if (includePfp) {
+            finalResponse.pfp = await getPfpUrl(identity.pdsUrl, identity.did);
+          }
+
+          const newCid = await computeCid(finalResponse);
+
+          if (clientCid && clientCid === newCid) {
+            return new Response(null, { status: 304, headers: corsHeaders() });
+          }
+
+          return new Response(JSON.stringify(finalResponse), {
+            status: 200,
+            headers: { ...corsHeaders(), "Content-Type": "application/json" },
+          });
+        } catch (error) {
+          const errorMessage =
+            error instanceof Error
+              ? error.message
+              : "An unknown error occurred";
+          return new Response(errorMessage, {
+            status: 500,
             headers: corsHeaders(),
           });
         }
-      } else if (req.method === "GET") {
-        const queryParam = url.searchParams.get("q");
-        if (!queryParam) {
-          return new Response("Missing 'q'", {
-            status: 400,
+      }
+
+      if (
+        pathname === "/xrpc/com.example.prototypeESQuery" ||
+        pathname === "/xrpc/party.whey.esav.esQuery"
+      ) {
+        let esQuery: Record<string, unknown>;
+        const clientCid: string | null = searchParams.get("cid");
+
+        if (req.method === "POST") {
+          try {
+            esQuery = await req.json();
+          } catch {
+            return new Response("Invalid JSON body", {
+              status: 400,
+              headers: corsHeaders(),
+            });
+          }
+        } else if (req.method === "GET") {
+          const queryParam = searchParams.get("q");
+          if (!queryParam)
+            return new Response("Missing 'q'", {
+              status: 400,
+              headers: corsHeaders(),
+            });
+
+          try {
+            esQuery = JSON.parse(queryParam);
+          } catch {
+            return new Response("Invalid JSON in 'q'", {
+              status: 400,
+              headers: corsHeaders(),
+            });
+          }
+        } else {
+          return new Response("Method not allowed", {
+            status: 405,
             headers: corsHeaders(),
           });
         }
 
         try {
-          esQuery = JSON.parse(queryParam);
-        } catch {
-          return new Response("Invalid JSON in 'q'", {
-            status: 400,
+          const esRes = await fetch(
+            `${config.es_url}/${config.index_name}/_search`,
+            {
+              method: "POST",
+              headers: { "Content-Type": "application/json" },
+              body: JSON.stringify(esQuery),
+            }
+          );
+
+          if (!esRes.ok) {
+            const errText = await esRes.text();
+            return new Response(`Elasticsearch error: ${errText}`, {
+              status: 500,
+              headers: corsHeaders(),
+            });
+          }
+
+          const result = await esRes.json();
+          const newCid = await computeCid(result);
+
+          if (clientCid && clientCid === newCid) {
+            return new Response(null, { status: 304, headers: corsHeaders() });
+          }
+
+          return new Response(JSON.stringify(result), {
+            status: 200,
+            headers: { ...corsHeaders(), "Content-Type": "application/json" },
+          });
+        } catch (error) {
+          const errorMessage =
+            error instanceof Error
+              ? error.message
+              : "An unknown error occurred";
+          return new Response(errorMessage, {
+            status: 500,
             headers: corsHeaders(),
           });
         }
-      } else {
-        return new Response("Method not allowed", {
-          status: 405,
-          headers: corsHeaders(),
-        });
       }
 
-      const esRes = await fetch(`${config.es_url}/${config.index_name}/_search`, {
+      return new Response("Not found", { status: 404, headers: corsHeaders() });
+    },
+    { port: Number(config.serve_port) }
+  );
+  console.log(`XRPC server running on http://localhost:${config.serve_port}`);
+}
+
+function cleanupClient(ws: WebSocket) {
+  console.log("WebSocket client disconnected. Cleaning up subscriptions.");
+  for (const [queryId, sub] of subscriptions.entries()) {
+    if (sub.clients.has(ws)) {
+      handleUnsubscribe(ws, queryId);
+      // sub.clients.delete(ws);
+      // if (sub.clients.size === 0) {
+      //   subscriptions.delete(queryId);
+      //   console.log(`Cleaned up empty subscription for queryId: ${queryId}`);
+      // }
+    }
+  }
+}
+
+function handleWebSocket(ws: WebSocket, config: AppConfig) {
+  console.log("WebSocket client connected.");
+  ws.isAlive = true;
+  
+  const pingInterval = setInterval(() => {
+    if (!ws.isAlive) {
+      console.log('Terminating dead WebSocket connection');
+      return ws.close();
+    }
+    ws.isAlive = false;
+    if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: 'ping' }));
+  }, 30000);
+
+  const onMessage = async (event: MessageEvent) => {
+    if (typeof event.data !== 'string') return;
+    
+    let msg;
+    try {
+      msg = JSON.parse(event.data);
+    } catch (_err) {
+      if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: 'error', error: 'Invalid JSON' }));
+      return;
+    }
+
+    switch (msg.type) {
+      case 'pong':
+        ws.isAlive = true;
+        break;
+      case 'subscribe':
+        if (msg.queryId && msg.esquery) {
+          await handleSubscribe(ws, msg.queryId, msg.esquery, config, msg.ecid);
+        } else {
+          ws.send(JSON.stringify({ type: 'error', error: 'Invalid subscribe message', request: msg }));
+        }
+        break;
+      case 'unsubscribe':
+        if (msg.queryId) {
+          handleUnsubscribe(ws, msg.queryId);
+        } else {
+          ws.send(JSON.stringify({ type: 'error', error: 'Invalid unsubscribe message', request: msg }));
+        }
+        break;
+      default:
+        ws.send(JSON.stringify({ type: 'error', error: 'Unknown message type', request: msg }));
+        break;
+    }
+  };
+
+  const onClose = () => { clearInterval(pingInterval); cleanupClient(ws); };
+  const onError = (e: Event) => { console.error('WebSocket error:', e); onClose(); };
+
+  ws.addEventListener('message', onMessage);
+  ws.addEventListener('close', onClose);
+  ws.addEventListener('error', onError);
+}
+
+async function handleSubscribe(
+  ws: WebSocket,
+  queryId: string,
+  esQuery: Record<string, unknown>,
+  config: AppConfig,
+  clientEcid?: string
+) {
+  try {
+    validateLiveQuery(esQuery);
+  } catch (err) {
+    console.error(`Invalid live query for queryId ${queryId}: ${err}`);
+    if (ws.readyState === WebSocket.OPEN) {
+      ws.send(JSON.stringify({
+        type: "error",
+        queryId,
+        error: `Invalid query for live subscription: ${err}`
+      }));
+    }
+    return;
+  }
+
+  let existingSub = subscriptions.get(queryId);
+
+  // a subscription (active or zombie) already exists
+  if (existingSub) {
+    // if the new query is different, the old state is invalid.
+    if (JSON.stringify(esQuery) !== JSON.stringify(existingSub.esQuery)) {
+        console.log(`[Sync] Query for ${queryId} has changed. Purging old state and re-running.`);
+        if (existingSub.zombieTimer) clearTimeout(existingSub.zombieTimer);
+        subscriptions.delete(queryId);
+        existingSub = undefined;
+    } else {
+        // if the query is the same. lets revive it if its a zombie
+        if (existingSub.zombieTimer) {
+          console.log(`[Zombie] Reviving zombie subscription: ${queryId}`);
+          clearTimeout(existingSub.zombieTimer);
+          delete existingSub.zombieTimer;
+        }
+        
+        existingSub.clients.add(ws);
+        
+        // only if the ecid is still salvageable
+        if (clientEcid && existingSub.ecid === clientEcid) {
+          console.log(`[Sync] Client for ${queryId} is already up-to-date. No-op.`);
+          return; 
+        }
+
+        console.log(`[Sync] Client for ${queryId} is out of sync. Re-running ES query.`);
+    }
+  }
+
+
+
+  // if (clientEcid && existingSub && existingSub.ecid === clientEcid) {
+  //   console.log(`Client re-subscribed to ${queryId} with matching ECID.`);
+  //   existingSub.clients.add(ws);
+  //   return;
+  // }
+
+  console.log(`Performing full sync for queryId: ${queryId}`);
+  try {
+    const esRes = await fetch(
+      `${config.es_url}/${config.index_name}/_search`,
+      {
         method: "POST",
         headers: { "Content-Type": "application/json" },
         body: JSON.stringify(esQuery),
+      }
+    );
+
+    if (!esRes.ok) throw new Error(`Elasticsearch error: ${await esRes.text()}`);
+
+    const esResult = await esRes.json();
+    const hits = esResult.hits?.hits ?? [];
+
+    const documents: Record<string, { cid: string, doc: Record<string, unknown> }> = {};
+    const newResultState: SubscriptionResult[] = [];
+
+    for (const hit of hits) {
+      const uri = hit._id as string;
+      const source = hit._source;
+      
+      newResultState.push({
+        uri: uri,
+        sortValues: hit.sort || [],
       });
 
-      if (!esRes.ok) {
-        const errText = await esRes.text();
-        return new Response(`Elasticsearch error: ${errText}`, {
-          status: 500,
-          headers: corsHeaders(),
-        });
-      }
+      documents[uri] = {
+        cid: source['$metadata.cid'],
+        doc: source,
+      };
+    }
+    
+    const resultUris = newResultState.map(item => item.uri);
+    const newEcid = await computeCid(resultUris);
 
-      const result = await esRes.json();
-      return new Response(JSON.stringify(result), {
-        status: 200,
-        headers: {
-          ...corsHeaders(),
-          "Content-Type": "application/json",
-        },
+    if (existingSub) {
+      existingSub.clients.add(ws);
+      existingSub.result = newResultState;
+      existingSub.ecid = newEcid;
+      existingSub.esQuery = esQuery;
+    } else {
+      subscriptions.set(queryId, {
+        esQuery,
+        clients: new Set([ws]),
+        ecid: newEcid,
+        result: newResultState,
       });
     }
 
-    return new Response("Not found", { status: 404, headers: corsHeaders() });
-  }, { port: Number(config.serve_port) });
+    ws.send(JSON.stringify({
+      type: "query-delta",
+      documents,
+      queries: {
+        [queryId]: {
+          ecid: newEcid,
+          result: resultUris,
+        }
+      }
+    }));
+
+  } catch (err) {
+    console.error(`Error subscribing client to query ${queryId}:`, err);
+    if (ws.readyState === WebSocket.OPEN) {
+        ws.send(JSON.stringify({ type: "error", queryId, error: `Subscription failed: ${err instanceof Error ? err.message : 'Unknown error'}` }));
+    }
+  }
 }
 
-function corsHeaders(): HeadersInit {
-  return {
-    "Access-Control-Allow-Origin": "*",
-    "Access-Control-Allow-Methods": "GET,POST,OPTIONS",
-    "Access-Control-Allow-Headers": "Content-Type",
-  };
-}
+function transitionToZombie(queryId: string) {
+  const sub = subscriptions.get(queryId);
+  if (!sub || sub.clients.size > 0) {
+    return;
+  }
+  
+  if (sub.zombieTimer) {
+    clearTimeout(sub.zombieTimer);
+  }
+
+  console.log(`[Zombie] Subscription ${queryId} has no clients. Entering zombie state for 5 minutes.`);
+
+  sub.zombieTimer = setTimeout(() => {
+    subscriptions.delete(queryId);
+    console.log(`[Zombie] Purged zombie subscription: ${queryId}`);
+  }, ZOMBIE_LIFETIME_MS);
+}
+
+function handleUnsubscribe(ws: WebSocket, queryId: string) {
+  const sub = subscriptions.get(queryId);
+  if (!sub) return;
+  sub.clients.delete(ws);
+  console.log(`[Sync] Client unsubscribed from ${queryId}. Remaining clients: ${sub.clients.size}`);
+
+  if (sub.clients.size === 0) {
+    transitionToZombie(queryId);
+  }
+  // if (sub.clients.size === 0) {
+  //   subscriptions.delete(queryId);
+  //   console.log(`Subscription removed for queryId: ${queryId} (last client disconnected)`);
+  // } else {
+  //   console.log(`Client unsubscribed from ${queryId}`);
+  // }
+}
+
+declare global {
+  interface WebSocket {
+    isAlive: boolean;
+    ping(): void;
+    terminate?(): void;
+  }
+}
+
+async function getServiceEndpointFromDid(
+  did: string
+): Promise<{ serviceEndpoint: string; handle?: string }> {
+  const didres = new DidResolver({});
+  const doc = await didres.resolve(did) as {
+    service?: Array<{
+      type: string;
+      serviceEndpoint: string;
+    }>;
+    alsoKnownAs?: string[];
+    [key: string]: unknown;
+  } | null;
+  if (!doc) {
+    throw new Error(`Could not resolve DID document for: ${did}`);
+  }
+  const endpointRaw = doc.service?.find(
+    (s: any) => s.type === "AtprotoPersonalDataServer"
+  )?.serviceEndpoint;
+  if (!endpointRaw) {
+    throw new Error(`Service endpoint not found for DID: ${did}`);
+  }
+  function getServiceEndpointString(endpoint: unknown): string {
+    if (typeof endpoint === "string") return endpoint;
+    if (
+      typeof endpoint === "object" &&
+      endpoint &&
+      "uri" in endpoint &&
+      typeof endpoint.uri === "string"
+    ) {
+      return endpoint.uri;
+    }
+    throw new Error("Unsupported serviceEndpoint format");
+  }
+  const serviceEndpoint = getServiceEndpointString(endpointRaw);
+  let handle: string | undefined;
+  if (Array.isArray(doc.alsoKnownAs) && doc.alsoKnownAs.length > 0) {
+    const aka = doc.alsoKnownAs[0];
+    if (typeof aka === "string" && aka.startsWith("at://")) {
+      handle = aka.slice("at://".length);
+    }
+  }
+  return { serviceEndpoint, handle };
+}
+
+async function resolveWithHandleOrDid({
+  handle,
+  did,
+}: {
+  handle?: string;
+  did?: string;
+}): Promise<{ did: string; pdsUrl: string; handle?: string }> {
+  if (did) {
+    const { serviceEndpoint, handle: resolvedHandle } =
+      await getServiceEndpointFromDid(did);
+    return {
+      did,
+      pdsUrl: serviceEndpoint,
+      handle: resolvedHandle,
+    };
+  } else if (handle) {
+    const hdlres = new HandleResolver();
+    const resolvedDid = await hdlres.resolve(handle);
+    if (!resolvedDid) {
+      throw new Error(`Could not resolve handle: ${handle}`);
+    }
+    const { serviceEndpoint } = await getServiceEndpointFromDid(resolvedDid);
+    return { did: resolvedDid, pdsUrl: serviceEndpoint, handle };
+  } else {
+    throw new Error("Either handle or did must be provided");
+  }
+}
+
+async function getPfpUrl(
+  pdsUrl: string,
+  did: string
+): Promise<string | undefined> {
+  try {
+    const profileUrl = `${pdsUrl}/xrpc/com.atproto.repo.getRecord?repo=${did}&collection=app.bsky.actor.profile&rkey=self`;
+    const response = await fetch(profileUrl);
+    if (!response.ok) {
+      console.warn(
+        `Failed to fetch profile for ${did} from ${pdsUrl}: ${response.status}`
+      );
+      return undefined;
+    }
+    const data = await response.json();
+    const cid = data?.value?.avatar?.ref?.["$link"];
+    const pfpurl = `${pdsUrl}/xrpc/com.atproto.sync.getBlob?did=${did}&cid=${cid}`;
+    return pfpurl;
+  } catch (error) {
+    console.error(`Error fetching PFP for ${did}:`, error);
+    return undefined;
+  }
+}