feat: poll hashtags, also consider hidden tags for keywords

(the latter is mostly so that hidden tags for uspol get considered)

This technically uses the Bluesky search endpoint, so please set
your poll interval quite high.

(On my bot, the uspol hashtag will apply a label to EVERY result!)

main.ts

···
 import { pollBlueskyList } from "./src/poll_list.ts";
 import { firehoseEventToRecordView } from "./src/convert.ts";
 import { RedisClient } from "jsr:@iuioiua/r2d2";
+import { pollHashtag } from "./src/poll_hashtag.ts";
 const { BskyAgent } = api; // RIP need module export
 
 // load .env file
···
   .option("--poll-feed <feedAtUri>", "A feed to check periodically. You'll need the feed's AT URI.", {collect: true})
   .option("--poll-chronological-feed <feedAtUri>", "A feed to check periodically. You'll need the feed's AT URI. Automatically paginates.", {collect: true})
   .option("--poll-list <listAtUri>", "A list to check periodically. This may point to an AT URI of a Bluesky list, or to a file where each line is the DID or handle of the account to crawl.", {collect: true, })
+  .option("--hashtag <hashtag>", "A hashtag (without the hash) or hidden tag to check periodically.", {collect: true, })
   .option("--firehose", "Listen on the firehose.")
   .option("--poll-reports", "Poll for reports.")
   //.option("--crawl-quote-chain", "Follow the quote chain.")
···
   .env("STAR_SPANGLED_PAGE_SIZE=<limit:number>", "For chronological feeds and lists, how large each page should be.", {prefix: "STAR_SPANGLED_"})
   .parse(Deno.args)
 
-if (!options.firehose && !options.pollReports && !options.pollFeed && !options.pollChronologicalFeed && !options.pollList) {
+if (!options.firehose && !options.pollReports && !options.pollFeed && !options.pollChronologicalFeed && !options.pollList && !options.hashtag) {
   console.error("Must specify at least one option")
   cmd.showHelp()
   Deno.exit(127)
···
     }
     for (const listAtUri of options.pollList || []) {
       await pollBlueskyList(labelerAgent, me, listAtUri)
+    }
+    for (const hashtag of options.hashtag || []) {
+      await pollHashtag(labelerAgent, me, hashtag)
     }
   } finally {
     console.debug(`-- Polling done. Will poll again in ${POLL_INTERVAL} seconds`)

src/handler.ts

···
  *  points. */
 export const CRAWLED_BONUS_POINTS = 10;
 
-const publicAgent = new Agent(new CredentialSession(new URL("https://public.api.bsky.app"), fetch, undefined));
+export const publicAgent = new Agent(new CredentialSession(new URL("https://public.api.bsky.app"), fetch, undefined));
 
 function escapeRegExp(string: string) {
     return string.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); // $& means the whole matched string
···
     const bridgyOriginalText = (post.record as any).bridgyOriginalText;
     allTextList.push(bridgyOriginalText)
     // TODO: if BridgyFed ever adds a CW field, I can add it here
+  }
+  for (const tag in (post.record as AppBskyFeedPost.Record).tags) {
+    allTextList.push("#"+tag);
   }
   const allText = allTextList.join(" ");
   const splitPost = allText.toLowerCase().split(WORD_SEPARATOR_PATTERN).filter((v,_,__) => v != "")

src/poll_hashtag.ts

···
+import { Agent, AppBskyActorGetProfile } from "@atproto/api";
+import { auditPostForKeywords, publicAgent } from "./handler.ts";
+import { DRY_RUN, INITIAL_PAGES_LIMIT, PAGE_SIZE, PAGES_LIMIT, redis, VERBOSE } from "../main.ts";
+import { retry } from "@atproto/common";
+import { hasAlreadyHandled } from "./redis.ts";
+import { clamp } from "./utils.ts";
+
+const hashtagMostRecent: Record<string, string> = {}
+async function setHashtagCursor(hashtag: string, cursor: string) {
+  if (redis) {
+    return await retry(async () => await redis!.sendCommand(["HSET", "hashtagCursors", hashtag, cursor]), {maxRetries: 3});
+  } else {
+    return hashtagMostRecent[hashtag] = cursor;
+  }
+}
+async function getHashtagCursor(hashtag: string): Promise<string|undefined> {
+  if (redis) {
+    const reply = await retry(async () => await redis!.sendCommand(["HGET", "hashtagCursors", hashtag]), {maxRetries: 3});
+    if (typeof reply == "string") return reply;
+    return undefined;
+  } else {
+    return hashtagMostRecent[hashtag];
+  }
+}
+
+export async function pollHashtag(agent: Agent, me: AppBskyActorGetProfile.Response, hashtag: string, cursor?: string|undefined, depth: number = 0) {
+  // deno-lint-ignore no-var
+  var anyPostFailed = false;
+  try {
+    const feedEntries = await agent.app.bsky.feed.searchPosts({
+      q: "#"+hashtag,
+      limit: clamp(PAGE_SIZE, 1, 100),
+      cursor: cursor || undefined,
+      sort: "latest"
+    }, {headers: {"atproto-proxy": ""}});
+    if (!feedEntries.success) return;
+    if (feedEntries.data.posts.length == 0) return;
+    const mostRecent = await getHashtagCursor(hashtag);
+    if (INITIAL_PAGES_LIMIT === 0 && !mostRecent) {
+      // can maybe be a zero-like value (undefined),
+      // so we have to strict type match.
+      // If INITIAL_PAGES_LIMIT is 0, don't do any initial labeling at all.
+      return;
+    }
+    for (const post of feedEntries.data.posts) {
+      if (post.uri == mostRecent) {
+        if (VERBOSE) console.log("Finishing with hashtag polling early")
+        if (depth == 0 && !anyPostFailed) {
+          if (VERBOSE) console.log("Most recent before:", mostRecent);
+          await setHashtagCursor(hashtag, feedEntries.data.posts[0].uri);
+          if (VERBOSE) console.log("Most recent now:", await getHashtagCursor(hashtag));
+        }
+        return; // out of pollFeed entirely
+      }
+      // // deno-lint-ignore no-explicit-any
+      // const record = post.post.record as any
+      // record.uri = post.post.uri;
+      // record.cid = post.post.cid;
+      try {
+        if (!mostRecent) {
+          const alreadyHandled = await hasAlreadyHandled(post.uri);
+          if (alreadyHandled) {
+            if (VERBOSE) console.log("Already handled this one:", post.uri);
+            continue;
+          } else if (alreadyHandled === null) {
+            const recResponse = await agent.tools.ozone.moderation.queryStatuses({
+              subject: post.uri,
+            });
+            if (recResponse.success && recResponse.data.subjectStatuses.length != 0) {
+              if (recResponse.data.subjectStatuses[0].tags?.includes("auto-handled") || recResponse.data.subjectStatuses[0].reviewState != "lex:tools.ozone.moderation.defs#reviewNone") {
+                continue; // skip this one, it's already been checked
+              }
+            }
+          }
+        }
+        if (VERBOSE || DRY_RUN) console.log(post.uri, `(#${hashtag})`);
+        try {
+          await auditPostForKeywords(post, agent, me, {})
+        } catch(e) {
+          console.error("Getting keywords for %s failed:", post.uri, e)
+          anyPostFailed = true;
+        }
+      } catch(e) {
+        console.error("Invalid post under hashtag:", post)
+        console.error("Cause:", e)
+        anyPostFailed = true;
+      }
+    }
+    if (depth == 0 && !anyPostFailed) {
+      if (VERBOSE) console.log("Most recent before:", mostRecent);
+      await setHashtagCursor(hashtag, feedEntries.data.posts[0].uri);
+      if (VERBOSE) console.log("Most recent now:", await getHashtagCursor(hashtag));
+    }
+    const limit = mostRecent ? PAGES_LIMIT : INITIAL_PAGES_LIMIT || PAGES_LIMIT
+    if (depth < limit && feedEntries.data.cursor) await pollHashtag(agent, me, hashtag, feedEntries.data.cursor, depth+1)
+  } catch(e) {
+    console.error("Error when polling #%s:", hashtag, e)
+  }
+}