Something went wrong. Try again.
watches bsky threads
Something went wrong. Try again.
4.6 kB · 105 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106import isEmpty from "just-is-empty";import { getRecordFeed } from "../consts";import { lookupName } from "../services/bskyNameLookup";import type { ParseThreadData } from "../types";
export async function parseThread(env: Env, ctx: ExecutionContext, thread: string, data: ParseThreadData, atCursor?: string) { const allRecords = await fetch(getRecordFeed(thread, atCursor), { headers: { "Accept": "application/json", "User-Agent": env.USER_AGENT }, });
if (allRecords.ok) { const jsonInfo: ATRecordBlob = await allRecords.json<ATRecordBlob>(); const totalRecords = jsonInfo.total;
// this can never be a string return, only null or data. const previousRecord: LandmarkData|null = data.threadData.get(thread) as LandmarkData|null; const hasPreviousRecord: boolean = (previousRecord !== null);
// if we have a record difference, it's time to dig deeper const hasCursor: boolean = !isEmpty(atCursor); const useCursor: string|null = (hasCursor) ? atCursor! : hasPreviousRecord ? previousRecord!.cursor : null;
// if we have a cursor and that cursor != the return's cursor // or if we have a previous record, no cursor, and previous record does not equate to return. if (jsonInfo.cursor !== null && useCursor !== null && useCursor !== jsonInfo.cursor) { console.log(`${thread} - traversing down using cursor: ${useCursor} got ${jsonInfo.cursor} to follow`); await parseThread(env, ctx, thread, data, jsonInfo.cursor); // And then we'll parse from here. }
// I have no records oh god help me please if (previousRecord === null || previousRecord.total <= totalRecords || useCursor != jsonInfo.cursor) { // check number of records in this new swarm if (jsonInfo.linking_records.length > 0) { // get the first record for comparison const firstRKey: string = jsonInfo.linking_records[0].rkey; // check to see if any records are different if (hasPreviousRecord && firstRKey == previousRecord!.last_top_record) { console.log(`${thread} - no new changes`); return; } const previousRecordCount = (previousRecord?.total ?? 0); const recordDelta = totalRecords - previousRecordCount;
// Go until we find a record we've already processed. for (const record of jsonInfo.linking_records) { // skip any messages that are written by me if (record.did === env.SKIP_DID) continue;
// if we have hit the last landmark we need to get out if (hasPreviousRecord && record.rkey === previousRecord!.last_top_record) { console.log(`${thread} - processed all posts up until last record, breaking out`); break; }
// check if we have reviewed this record before const recordExists: string|null = await env.KV.get(record.rkey); if (recordExists !== null) { if (hasCursor) { console.log(`${thread} - record ${record.rkey} exists, with cursor ${useCursor}, breaking out`); break; } else { console.log(`${thread} - record ${record.rkey} exists, skipping`); continue; } }
// We have never seen this object in our life, ever. // "I know this, and I love you." const newRecordTask: BSkyRecordTask = { recordNumber: data.current_total + 1, username: await lookupName(record.did), did: record.did, rkey: record.rkey, thread: thread, recurseDepth: 0 };
// push to the queue await env.THREAD_UPDATE_QUEUE.send(newRecordTask, {contentType: 'v8'}) ++data.current_total; } // Sub-traversals should not write landmark data, so only write if we are main thread. if (atCursor === undefined) { console.log(`${thread} - new landmark created, ${firstRKey}! Processed ${recordDelta} records`);
const updatedKVRecord: LandmarkData = { cursor: jsonInfo.cursor ?? "", total: jsonInfo.total, last_top_record: firstRKey }; await env.KV.put(thread, JSON.stringify(updatedKVRecord)); }
} else { console.warn(`${thread} - NO RECORDS EXIST.`); } } else { console.log(`${thread} - detected no changes in record`); } } else { console.warn(`${thread} - Unable to get current bsky records, got return: "${allRecords.status}"`); }}