pds-feedgen / feedgen.ts

import {
  AppBskyFeedDescribeFeedGenerator,
  AppBskyFeedGetFeedSkeleton,
} from "@atcute/bluesky";
import {
  CompositeDidDocumentResolver,
  PlcDidDocumentResolver,
  WebDidDocumentResolver,
} from "@atcute/identity-resolver";
import {
  parseResourceUri,
  type Nsid,
  type ResourceUri,
} from "@atcute/lexicons/syntax";
import {
  AuthRequiredError,
  InvalidRequestError,
  XRPCRouter,
  json,
} from "@atcute/xrpc-server";
import { ServiceJwtVerifier, type VerifiedJwt } from "@atcute/xrpc-server/auth";
import { cors } from "@atcute/xrpc-server/middlewares/cors";
import type { Statement } from "@db/sqlite";

import { db } from "./common/db.ts";
import type { Author, DID, Post } from "./common/types.ts";

const publisher = Deno.env.get("PUBLISHER") ?? "did:example:bob";
const hostname = Deno.env.get("HOSTNAME");
if (!hostname) {
  console.error("HOSTNAME not provided! Exiting now.");
  Deno.exit(1);
}

const baseDID: DID = `did:web:${hostname}`;

const app = new XRPCRouter({ middlewares: [cors()] });
const didResolver = new CompositeDidDocumentResolver({
  methods: {
    plc: new PlcDidDocumentResolver(),
    web: new WebDidDocumentResolver(),
  },
});
const verifier = new ServiceJwtVerifier({
  serviceDid: baseDID,
  resolver: didResolver,
});

const feeds: Record<
  string,
  { default: Statement; cursor: Statement; pds: boolean }
> = {
  "your-pds": {
    default: db.prepare(
      `SELECT a.uri, a.indexed_at FROM posts a
        INNER JOIN authors b ON a.author = b.did
        WHERE b.pds = ?1
        ORDER BY a.indexed_at DESC, a.cid DESC LIMIT ?2;`
    ),
    cursor: db.prepare(
      `SELECT a.uri, a.indexed_at FROM posts a
        INNER JOIN authors b ON a.author = b.did
        WHERE b.pds = ?1
        AND a.indexed_at < ?2
        ORDER BY a.indexed_at DESC, a.cid DESC LIMIT ?3;`
    ),
    pds: true,
  },
  "non-bsky-pds": {
    default: db.prepare(
      `SELECT a.uri, a.indexed_at FROM posts a
        INNER JOIN authors b ON a.author = b.did
        WHERE b.pds_base != 'bsky.network'
        AND b.pds_base != 'brid.gy'
        ORDER BY a.indexed_at DESC, a.cid DESC LIMIT ?1;`
    ),
    cursor: db.prepare(
      `SELECT a.uri, a.indexed_at FROM posts a
        INNER JOIN authors b ON a.author = b.did
        WHERE b.pds_base != 'bsky.network'
        AND b.pds_base != 'brid.gy'
        AND a.indexed_at < ?1
        ORDER BY a.indexed_at DESC, a.cid DESC LIMIT ?2;`
    ),
    pds: false,
  },
};

const requireAuth = async (
  request: Request,
  lxm: Nsid
): Promise<VerifiedJwt> => {
  const auth = request.headers.get("authorization");
  if (auth === null) {
    throw new AuthRequiredError({
      description: `missing authorization header`,
    });
  }
  if (!auth.startsWith("Bearer ")) {
    throw new AuthRequiredError({
      description: `invalid authorization scheme`,
    });
  }

  const jwtString = auth.slice("Bearer ".length).trim();

  const result = await verifier.verify(jwtString, { lxm });
  if (!result.ok) {
    throw new AuthRequiredError(result.error);
  }

  return result.value;
};

const getAuthor = db.prepare("SELECT pds FROM authors WHERE did = ?");

app.add(AppBskyFeedGetFeedSkeleton.mainSchema, {
  async handler({ request, params: { feed, limit, cursor } }) {
    const feedUri = parseResourceUri(feed);
    if (!feedUri.ok || !feedUri.value.rkey) {
      throw new InvalidRequestError();
    }

    const feedQuery = feeds[feedUri.value.rkey];

    if (
      feedUri.value.repo !== publisher ||
      feedUri.value.collection !== "app.bsky.feed.generator" ||
      !feedQuery
    ) {
      throw new InvalidRequestError({
        error: "UnsupportedAlgorithm",
        description: "Unsupported algorithm",
      });
    }

    let pds = "";

    if (feedQuery.pds) {
      const jwt = await requireAuth(request, "app.bsky.feed.getFeedSkeleton");

      const author = getAuthor.get<Author>(jwt.issuer);
      if (author) {
        pds = author.pds;
      } else {
        const resolved = await didResolver.resolve(jwt.issuer as DID);
        for (const service of resolved.service ?? []) {
          if (
            service.type == "AtprotoPersonalDataServer" &&
            typeof service.serviceEndpoint === "string"
          ) {
            pds = service.serviceEndpoint;
          }
        }
        if (typeof pds !== "string")
          throw new InvalidRequestError({
            error: "NoServiceEndpoint",
            description: "No service endpoint",
          });
      }
    }

    let res: Post[];
    if (cursor) {
      const timeStr = new Date(parseInt(cursor, 10)).toISOString();
      if (feedQuery.pds) {
        res = feedQuery.cursor.all(pds, timeStr, limit);
      } else {
        res = feedQuery.cursor.all(timeStr, limit);
      }
    } else {
      if (feedQuery.pds) {
        res = feedQuery.default.all(pds, limit);
      } else {
        res = feedQuery.default.all(limit);
      }
    }

    const posts = res.map((row) => ({
      post: row.uri,
    }));

    let cs: string | undefined;
    const last = res.at(-1);
    if (last) {
      cs = new Date(last.indexed_at).getTime().toString(10);
    }

    return json({
      cursor: cs,
      feed: posts,
    });
  },
});

app.add(AppBskyFeedDescribeFeedGenerator.mainSchema, {
  handler() {
    const feedArray = Object.keys(feeds).map((v) => ({
      uri: `at://${publisher}/app.bsky.feed.generator/${v}` as ResourceUri,
    }));
    return json({
      did: baseDID,
      feeds: feedArray,
    });
  },
});

export default app;