diff --git a/apps/web/tailwind.config.js b/apps/web/tailwind.config.js index ad09853..d0d5069 100644 --- a/apps/web/tailwind.config.js +++ b/apps/web/tailwind.config.js @@ -1,6 +1,6 @@ /** @type {import('tailwindcss').Config} */ export default { - content: ['./src/**/*.{astro,html,js,jsx,md,mdx,svelte,ts,tsx,vue}'], + content: ['./src/**/*.{astro,html,js,jsx,md,mdx,svelte,ts,tsx,vue}', '../../node_modules/react-bluesky-embed/**/*.js'], theme: { extend: {}, }, diff --git a/packages/atproto/domain/get-tech-all-feed.ts b/packages/atproto/domain/get-tech-all-feed.ts index 099196e..b0110f6 100644 --- a/packages/atproto/domain/get-tech-all-feed.ts +++ b/packages/atproto/domain/get-tech-all-feed.ts @@ -4,7 +4,8 @@ import { followTable, postScores, postTable } from "../db/schema"; export async function getTechAllFeed( ctx: AtContext, - did: string + did: string, + limit: number = 30 ): Promise> { const fls = ctx.db .select() @@ -21,7 +22,7 @@ export async function getTechAllFeed( ) .where(gte(postScores.avgScore, 80)) // TODO: Raise this to 80 .orderBy(desc(postTable.created)) - .limit(50); + .limit(limit); return posts.map((post) => post.id); } diff --git a/packages/atproto/domain/jetstream.ts b/packages/atproto/domain/jetstream.ts index a04a8c8..4bd1cb9 100644 --- a/packages/atproto/domain/jetstream.ts +++ b/packages/atproto/domain/jetstream.ts @@ -38,22 +38,22 @@ export const JETSTREAM_BASE_URL = // TODO: Let each listener have id, and cursor. Then persist it let subscriptions: Array | undefined; -export async function createJetStreamListener({ - wantedCollections = ["app.bsky.feed.post"], - wantedDids = [], - cursor, - onPostCreated, -}: { - wantedCollections?: Array; - wantedDids?: Array; +export async function createJetStreamListener(args: { + wantedCollections?: Readonly>; + wantedDids?: Readonly>; /** Typically the unixtime of the last received post */ - cursor?: string; + cursor?: Readonly; onPostCreated?: (post: FeedPostWithUri) => void; }) { + let wantedDids = args.wantedDids ?? []; + let wantedCollections = args.wantedCollections ?? []; + let cursor = args.cursor; + const { onPostCreated } = args; + async function init() { subscriptions = []; - await zstd.init(); - let remainingDids = wantedDids; + let remainingDids = [...wantedDids]; + console.log(`[jetstream] init ${wantedDids.length} dids requested`); while (remainingDids.length > 0) { const requestDids = remainingDids.splice( 0, @@ -61,9 +61,6 @@ export async function createJetStreamListener({ ? remainingDids.length : MAX_DIDS_PER_SOCKET ); - const wcQ = wantedCollections.map((c) => `wantedCollections=${c}`); - - const wdQ = requestDids.map((did) => `wantedDids=${did}`); // Requesting everything from the start crashes the socket const url = new URL(JETSTREAM_BASE_URL); @@ -132,16 +129,20 @@ export async function createJetStreamListener({ if (args.wantedCollections) wantedCollections = args.wantedCollections; if (args.wantedDids) wantedDids = args.wantedDids; if (args.cursor) cursor = args.cursor; - if (!subscriptions) return; + if (!subscriptions) { + return; + } // Close existing sockets and re-init for (const sub of subscriptions) { sub.close(); } + // TODO: this is broken await init(); } const zDictionary = Bun.file(path.join(__dirname, "../zstd_dictionary.dat")); + await zstd.init(); await init(); return { updateRequest }; } diff --git a/packages/atproto/domain/queue-for-classification.ts b/packages/atproto/domain/queue-for-classification.ts index d25452f..2abcea8 100644 --- a/packages/atproto/domain/queue-for-classification.ts +++ b/packages/atproto/domain/queue-for-classification.ts @@ -20,9 +20,13 @@ export async function queueForClassification( const langProb = lande(text); // If english isn't the most likely language, we skip it entirely - const isNotEnglish = - text.length > 20 && langProb[0] && langProb[0][0] !== "eng"; - if (isNotEnglish) { + const detectedNonEnglish = (() => { + if (text.length < 20) return false; + const [firstLang, pb] = langProb[0]; + if (firstLang !== "eng" && pb > 0.8) return true; + return false; // We just don't know + })(); + if (detectedNonEnglish) { console.log("[queue] not english: ", text); return; } diff --git a/packages/atproto/domain/subscription-service.ts b/packages/atproto/domain/subscription-service.ts deleted file mode 100644 index 0210098..0000000 --- a/packages/atproto/domain/subscription-service.ts +++ /dev/null @@ -1,129 +0,0 @@ -import type { AppBskyFeedPost } from "@atproto/api"; -import { Jetstream } from "@skyware/jetstream"; -import type { AtContext } from "../context"; -import { followTable } from "../db/schema"; -import { queueForClassification } from "./queue-for-classification"; -type PostRecord = AppBskyFeedPost.Record & { - reply?: { - parent?: { - cid: string; - uri: string; - }; - root?: { - cid: string; - uri: string; - }; - }; -}; - -type StreamCollection = { - did: string; - time_us: number; - kind: "commit"; - commit: { - collection: "app.bsky.feed.post"; - rev: string; - operation: "create"; - rkey: string; - record: PostRecord; - }; -}; - -export class PostSubscription { - socket: WebSocket | undefined = undefined; - jetstream: Jetstream | undefined = undefined; - ctx: AtContext; - cursor?: string; // TODO - constructor(ctx: AtContext) { - this.ctx = ctx; - } - - // TODO: Remove or retry. Not possible due to: https://github.com/oven-sh/bun/issues/8721 - public async listenJetstream() { - const res = await this.insterestingAccounts(); - if (!res) { - console.warn("No accounts returned, not listening for updates"); - return; - } - const wantedDids = res.map((r) => r.did); - - const jetstream = new Jetstream({ - wantedCollections: ["app.bsky.feed.post"], - wantedDids: wantedDids, - }); - - jetstream.onCreate("app.bsky.feed.post", (event) => { - console.log(`New post: ${event.commit.record.text}`); - }); - jetstream.on("error", (err) => { - console.error(err); - }); - jetstream.on("close", () => { - console.log("jetstream closed"); - }); - - jetstream.start(); - - this.jetstream = jetstream; - } - public async listen() { - const wantedDids = await this.insterestingAccounts(); - if (!wantedDids || wantedDids.length === 0) { - console.warn("No accounts returned, not listening for updates"); - return; - } - - const wantedDidsQuery = wantedDids - .map((r) => `wantedDids=${r.did}`) - .join("&"); - - const url = `wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post&${wantedDidsQuery}`; - console.log(`Opening wss to: ${url}`); - - const encodedUrl = encodeURI(url); - const wss = new WebSocket(encodedUrl); - wss.addEventListener("error", () => { - console.log("wss error"); - }); - wss.addEventListener("close", (ev) => { - console.log("wss socket closed", ev.code, ev.reason, ev.wasClean); - }); - wss.addEventListener("open", () => { - console.log("wss open"); - }); - - wss.addEventListener("message", async (ev) => { - const raw = ev.data as string; - const data = JSON.parse(raw) as StreamCollection; - console.log("message", data.kind, data.did); - // TODO: Filtering should be done by classifying pipeline - if (data.commit.operation !== "create") return; - if (data.commit.collection !== "app.bsky.feed.post") return; - if (!data.commit.record.langs?.includes("en")) return; - //if (data.commit.record.reply) return; - - await queueForClassification(this.ctx, { - createdAt: data.commit.record.createdAt, - text: data.commit.record.text, - uri: `at://${data.did}/app.bsky.feed.post/${data.commit.rkey}`, - authorId: data.did, - }); - - console.log("valid post", data); - }); - - this.socket = wss; - } - - private async insterestingAccounts() { - const res = await this.ctx.db - .selectDistinct({ did: followTable.follows }) - .from(followTable) - .limit(10); // This shouldn't be necessary! - return res; - } -} - -// We need a post subscription that stores the cursor and resumes -// If we hit the limit, then repeat until we dont -// Probably need this to monitor likes too