add blue sky fetcher

bruno/MicroBlogFetchders/Get Pixelfed Feed.bru

@@ -5,13 +5,14 @@ meta {
 }
 
 get {
-  url: https://gram.social/api/pixelfed/v1/accounts/703621281309160235/statuses?limit=9&only_media=true&min_id=1
+  url: https://gram.social/api/pixelfed/v1/accounts/703621281309160235/statuses?limit=5&only_media=true
   body: none
   auth: none
 }
 
 query {
-  limit: 9
+  limit: 5
   only_media: true
-  min_id: 1
+  ~min_id: 704018331666201874
+  ~max_id: 704018331666201874
 }

serverless.yml

@@ -12,9 +12,15 @@ provider:
   environment:
     POCKET_BASE_PW: ${env:POCKET_BASE_PW}
     POCKET_BASE_USER: ${env:POCKET_BASE_USER}
+    BLUE_SKY_API_KEY: ${env:BLUE_SKY_API_KEY}
+    BLUE_SKY_USERNAME: ${env:BLUE_SKY_USERNAME}
 
 functions:
   pixelfed:
     handler: src/pixelfed.run
     events:
       - schedule: rate(1 hour)
+  bluesky:
+    handler: src/bluesky.run
+    events:
+      - schedule: rate(1 hour)

src/bluesky.ts

@@ -0,0 +1,159 @@
+import pino from "pino";
+import { lambdaRequestTracker, pinoLambdaDestination } from "pino-lambda";
+import { MicroBlogBackend } from "./pocketbase";
+
+// logger
+const destination = pinoLambdaDestination();
+const logger = pino({}, destination);
+const withRequest = lambdaRequestTracker();
+
+// pocketbase
+const pb = new MicroBlogBackend(logger);
+
+type Session = {
+  did: string;
+  accessJwt: string;
+  refreshJwt: string;
+};
+
+type BlueSkyPost = {
+  cid: string;
+  embed?: {
+    images: {
+      fullsize: string;
+      alt: string;
+    }[];
+  };
+  record: {
+    createdAt: string; // '2024-06-25T05:32:06.269Z',
+    // embed: { '$type': 'app.bsky.embed.images', images: [Array] },
+    // facets: [ [Object] ],
+    text: string;
+    facets: {
+      features: {
+        $type: string; //"app.bsky.richtext.facet#tag",\n' +
+        tag?: string; //  "cooking"\n' +
+      }[];
+    }[];
+  };
+};
+
+const createSession = async (): Promise<Session> => {
+  const identifier = process.env.BLUE_SKY_USERNAME;
+  const apiKey = process.env.BLUE_SKY_API_KEY;
+  const body = JSON.stringify({ identifier, password: apiKey });
+  const url = "https://bsky.social/xrpc/com.atproto.server.createSession";
+  const res = await fetch(url, {
+    method: "POST",
+    headers: {
+      "Content-Type": "application/json",
+    },
+    body: body,
+  });
+  const data = (await res.json()) as Session;
+  return data;
+};
+
+const limit = 10;
+const getPostsUntilID = async (
+  session: Session,
+  id: string,
+  cursor: string | null = null,
+  oldFeed: BlueSkyPost[] = []
+): Promise<BlueSkyPost[]> => {
+  const params = new URLSearchParams();
+  params.append("actor", session.did);
+  params.append("limit", limit.toString());
+  if (cursor) {
+    params.append("cursor", cursor);
+  }
+
+  const urlWithParams = new URL(
+    "https://bsky.social/xrpc/app.bsky.feed.getAuthorFeed"
+  );
+  urlWithParams.search = params.toString();
+
+  const res = await fetch(urlWithParams, {
+    headers: {
+      Accept: "application/json",
+      Authorization: `Bearer ${session.accessJwt}`,
+    },
+  });
+  const rawData = (await res.json()) as {
+    feed: { post: BlueSkyPost }[];
+    cursor: string | null;
+  };
+  const rawFeed = rawData.feed;
+  const feed = rawFeed.map((item) => item.post);
+  cursor = rawData?.cursor;
+  const filteredFeed = [];
+  for (const post of feed) {
+    if (post.cid === id) {
+      break;
+    }
+    filteredFeed.push(post);
+  }
+
+  //  the post id we are searching until is in the res so return
+  if (filteredFeed.length !== feed.length) {
+    return filteredFeed.concat(oldFeed);
+  }
+  // there are more posts to add before the id
+  if (feed.length === limit) {
+    return getPostsUntilID(session, id, cursor, oldFeed.concat(feed));
+  }
+
+  // the id was not found in the feed, return everything
+  return oldFeed.concat(feed);
+};
+
+const savePost = async (post: BlueSkyPost) => {
+  const postData = {
+    remoteId: post.cid,
+    posted: post.record.createdAt,
+    source: "blue_sky" as const,
+    fullPost: post,
+    authorId: "travisshears.bsky.social",
+  };
+  return await pb.savePost(postData);
+};
+
+const saveTags = async (post: BlueSkyPost, postId: string) => {
+  for (const facet of post.record.facets) {
+    for (const feature of facet.features) {
+      if (feature.$type === "app.bsky.richtext.facet#tag") {
+        const tag = feature.tag;
+        if (tag) {
+          await pb.setTag(tag, postId);
+        }
+      }
+    }
+  }
+};
+
+const saveImages = async (post: BlueSkyPost, postId: string) => {
+  const images = post.embed?.images ?? [];
+  for (const image of images) {
+    await pb.saveAndSetImage(
+      { remoteURL: image.fullsize, alt: image.alt },
+      postId
+    );
+  }
+};
+
+exports.run = async (event: any, context: any) => {
+  withRequest(event, context);
+  const session = await createSession();
+  const lastSavedPostId = await pb.getLatestPostRemoteIDBySource("blue_sky");
+  const posts = await getPostsUntilID(session, lastSavedPostId ?? "");
+  posts.reverse(); // save the oldest post first so if we fail posts are not lost on the next run
+  if (posts.length === 0) {
+    logger.info("No new posts to save");
+  }
+  for (const post of posts) {
+    logger.info({ post }, "saving post");
+    const savedNewPost = await savePost(post);
+    await saveTags(post, savedNewPost.id);
+    await saveImages(post, savedNewPost.id);
+  }
+};

src/pixelfed.ts

@@ -96,7 +96,7 @@ const saveImages = async (post: PixelFedPost, postId: string) => {
 
 exports.run = async (event: any, context: any) => {
   withRequest(event, context);
-  const lastSavedPostId = await pb.getLatestPostId("pixelfed");
+  const lastSavedPostId = await pb.getLatestPostRemoteIDBySource("pixelfed");
   const posts = await getPostUntilId({ lastSavedId: lastSavedPostId });
   for (const post of posts) {
     logger.info({ post }, "saving post");

src/pocketbase.ts

@@ -64,15 +64,15 @@ export class MicroBlogBackend {
     }
   }
 
-  public async getLatestPostId(
+  public async getLatestPostBySource(
     postSource: MicroBlogPostSource
-  ): Promise<string | undefined> {
+  ): Promise<MicroBlogPost | undefined> {
     await this.checkLogin();
     try {
       const post = await this.pb
         .collection<MicroBlogPost>("micro_blog_posts")
         .getFirstListItem(`source = '${postSource}'`, { sort: "-posted" });
-      return post.id;
+      return post;
     } catch (error: any) {
       if (error.status === 404) {
         return undefined;
@@ -81,6 +81,11 @@ export class MicroBlogBackend {
     }
   }
 
+  public async getLatestPostRemoteIDBySource(postSource: MicroBlogPostSource) {
+    const post = await this.getLatestPostBySource(postSource);
+    return post?.remoteId;
+  }
+
   async getTag(tag: string): Promise<MicroBlogPostTag | undefined> {
     await this.checkLogin();
     try {