initial

.gitignore

···
+deno.lock

config.json

···
+{
+  "jetstream_url": "wss://jetstream2.us-west.bsky.network/subscribe",
+  "es_url": "http://localhost:9200",
+  "record_types": ["xyz.statusphere.status", "com.example.test.esav"],
+  "index_name": "appview-poc",
+  "serve_port": "8370",
+  "index_fields": {
+    "xyz.statusphere.status": {
+      "status": {
+        "id": "status",
+        "type": "text"
+      }
+    },
+    "com.example.test.esav": {
+      "text": {
+        "id": "text",
+        "type": "text"
+      },
+      "reply.parent.uri": {
+        "id": "reply.parent.uri",
+        "type": "keyword"
+      },
+      "reply.root.uri": {
+        "id": "reply.root.uri",
+        "type": "keyword"
+      }
+    },
+    "app.bsky.feed.post": {
+      "text": {
+        "id": "text",
+        "type": "text"
+      },
+      "reply.parent.uri": {
+        "id": "reply.parent.uri",
+        "type": "keyword"
+      },
+      "reply.root.uri": {
+        "id": "reply.root.uri",
+        "type": "keyword"
+      }
+    },
+    "app.bsky.feed.like": {
+      "subject.uri":{
+        "id": "subject.uri",
+        "type": "keyword"
+      }
+    }
+  }
+}
+
+

deno.json

···
+{
+  "tasks": {
+    "dev": "deno run -A ./src/main.ts"
+  }
+}

readme.md

···
+# Elastic Search AppView
+terrible i know right lmao
+
+send ES queries to `/xrpc/com.example.prototypeESQuery` either POST or GET q=
+
+change config in `config.json`
+
+run `deno task dev` to start
+
+you need elasticsearch configured and running first though, good luck with that

src/config.ts

···
+export interface AppConfig {
+  jetstream_url: string;
+  es_url: string;
+  record_types: string[];
+  index_name: string;
+  serve_port: string;
+  index_fields: Record<
+    string,
+    Record<
+      string,
+      {
+        id: string;
+        type: "text" | "keyword";
+      }
+    >
+  >;
+}
+
+export async function readConfig(path: string): Promise<AppConfig> {
+  const text = await Deno.readTextFile(path);
+  return JSON.parse(text);
+}

src/firehose.ts

···
+import { AppConfig } from "./config.ts";
+import { indexDocument, deleteDocument } from "./indexer.ts";
+
+export async function startFirehose(config: AppConfig) {
+  const ws = new WebSocket(config.jetstream_url);
+
+  ws.onopen = () => {
+    console.log("Connected to Jetstream");
+  };
+
+  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;
+
+    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 (!config.record_types.includes(collection)) return;
+
+    if (opType === "create" || opType === "edit") {
+      await indexDocument(config, aturi, record, cid, indexedAt);
+    } else if (opType === "delete") {
+      await deleteDocument(config, aturi);
+    }
+  };
+
+  ws.onerror = (e) => {
+    console.error("Jetstream error:", e);
+  };
+}

src/indexer.ts

···
+import { AppConfig } from "./config.ts";
+
+function atUriParts(uri: string) {
+  const parts = uri.split("/");
+  return {
+    did: parts[2],
+    collection: parts[3],
+    rkey: parts[4],
+  };
+}
+
+export async function ensureIndexMapping(config: AppConfig) {
+  const properties: Record<string, any> = {};
+
+  for (const fields of Object.values(config.index_fields)) {
+    for (const { id, type } of Object.values(fields)) {
+      properties[id] = {
+        type: type === "text" ? "text" : "keyword"
+      };
+    }
+  }
+
+  for (const key of [
+    "$metadata.uri",
+    "$metadata.cid",
+    "$metadata.indexedAt",
+    "$metadata.did",
+    "$metadata.collection",
+    "$metadata.rkey",
+  ]) {
+    properties[key] = { type: "keyword" };
+  }
+
+  await fetch(`${config.es_url}/${config.index_name}`, {
+    method: "PUT",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({
+      mappings: { properties }
+    }),
+  });
+}
+
+function extractIndexedFields(
+  record: any,
+  indexSpec: Record<string, { id: string; type: string }>
+): Record<string, unknown> {
+  const result: Record<string, unknown> = {};
+
+  for (const [jsonPath, { id }] of Object.entries(indexSpec)) {
+    const pathSegments = jsonPath.split(".");
+    let value = record;
+    for (const segment of pathSegments) {
+      if (typeof value !== "object" || value === null) {
+        value = undefined;
+        break;
+      }
+      value = value[segment];
+    }
+
+    if (value !== undefined) {
+      result[id] = value;
+    }
+  }
+
+  return result;
+}
+
+export async function indexDocument(
+  config: AppConfig,
+  uri: string,
+  record: any,
+  cid: string,
+  indexedAt: string
+) {
+  const { did, collection, rkey } = atUriParts(uri);
+  const recordType = collection;
+
+  const indexSpec = config.index_fields?.[recordType];
+  if (!indexSpec) return; // skip if record type not configured
+
+  const indexedFields = extractIndexedFields(record, indexSpec);
+
+  const metadata = {
+    "$metadata.uri": uri,
+    "$metadata.cid": cid,
+    "$metadata.indexedAt": indexedAt,
+    "$metadata.did": did,
+    "$metadata.collection": collection,
+    "$metadata.rkey": rkey,
+  };
+
+  const body = {
+    ...metadata,
+    ...indexedFields,
+    "$raw": record,
+  };
+
+  const res = await fetch(
+    `${config.es_url}/${config.index_name}/_doc/${encodeURIComponent(uri)}`,
+    {
+      method: "PUT",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify(body),
+    }
+  );
+
+  if (!res.ok) {
+    console.error("Indexing failed:", await res.text());
+  }
+}
+
+export async function deleteDocument(config: AppConfig, uri: string) {
+  const res = await fetch(
+    `${config.es_url}/${config.index_name}/_doc/${encodeURIComponent(uri)}`,
+    {
+      method: "DELETE",
+    }
+  );
+
+  if (!res.ok && res.status !== 404) {
+    console.error("Delete failed:", await res.text());
+  }
+}

src/main.ts

···
+import { readConfig } from "./config.ts";
+import { startFirehose } from "./firehose.ts";
+import { ensureIndexMapping } from "./indexer.ts";
+import { setupXRPCServer } from "./xrpc.ts";
+
+const config = await readConfig("./config.json");
+
+// prepare indexes
+ensureIndexMapping(config);
+
+startFirehose(config);
+
+setupXRPCServer(config);

src/xrpc.ts

···
+import { AppConfig } from "../src/config.ts";
+import { serve } from "https://deno.land/std/http/server.ts";
+
+export function setupXRPCServer(config: AppConfig) {
+  serve(async (req) => {
+    const url = new URL(req.url);
+
+    if (url.pathname === "/xrpc/com.example.prototypeESQuery") {
+      let esQuery: any;
+
+      if (req.method === "POST") {
+        try {
+          esQuery = await req.json();
+        } catch {
+          return new Response("Invalid JSON body", { status: 400 });
+        }
+      } else if (req.method === "GET") {
+        const queryParam = url.searchParams.get("q");
+        if (!queryParam) {
+          return new Response("Missing 'q'", { status: 400 });
+        }
+
+        try {
+          esQuery = JSON.parse(queryParam);
+        } catch {
+          return new Response("Invalid JSON in 'q'", { status: 400 });
+        }
+      } else {
+        return new Response("Method not allowed", { status: 405 });
+      }
+
+      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 });
+      }
+
+      const result = await esRes.json();
+      return Response.json(result.hits.hits.map((hit: any) => hit._source));
+    }
+
+    return new Response("Not found", { status: 404 });
+  }, { port: Number(config.serve_port) });
+}