Rebase blacsky wth the main line

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Bluesky PBLLC
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

src/config.ts

@@ -9,8 +9,11 @@ export type AppContext = {
 
 export type Config = {
   port: number
+  listenhost: string
   hostname: string
   dblocation: string
   subscriptionEndpoint: string
   serviceDid: string
+  publisherDid: string
+  subscriptionReconnectDelay: number
 }

src/index.ts

@@ -7,17 +7,22 @@ const run = async () => {
   const serviceDid =
     maybeStr(process.env.FEEDGEN_SERVICE_DID) ?? `did:web:${hostname}`
   const server = FeedGenerator.create({
-    port: maybeInt(process.env.PORT) ?? 3000,
+    port: maybeInt(process.env.FEEDGEN_PORT) ?? 3000,
+    listenhost: maybeStr(process.env.FEEDGEN_LISTENHOST) ?? 'localhost',
     dblocation: maybeStr(process.env.FEEDGEN_DB_LOCATION) ?? ':memory:',
     subscriptionEndpoint:
       maybeStr(process.env.FEEDGEN_SUBSCRIPTION_ENDPOINT) ??
       'wss://bsky.social',
+    publisherDid:
+      maybeStr(process.env.FEEDGEN_PUBLISHER_DID) ?? 'did:example:alice',
+    subscriptionReconnectDelay:
+      maybeInt(process.env.FEEDGEN_SUBSCRIPTION_RECONNECT_DELAY) ?? 3000,
     hostname,
     serviceDid,
   })
   await server.start()
   console.log(
-    `🤖 running feed generator at http://localhost:${server.cfg.port}`,
+    `🤖 running feed generator at http://${server.cfg.listenhost}:${server.cfg.port}`,
   )
 }
 

src/methods/describe-generator.ts

@@ -1,10 +1,17 @@
 import { Server } from '../lexicon'
 import { AppContext } from '../config'
 import algos from '../algos'
+import { AtUri } from '@atproto/uri'
 
 export default function (server: Server, ctx: AppContext) {
   server.app.bsky.feed.describeFeedGenerator(async () => {
-    const feeds = Object.keys(algos).map((uri) => ({ uri }))
+    const feeds = Object.keys(algos).map((shortname) => ({
+      uri: AtUri.make(
+        ctx.cfg.publisherDid,
+        'app.bsky.feed.generator',
+        shortname,
+      ).toString(),
+    }))
     return {
       encoding: 'application/json',
       body: {

src/methods/feed-generation.ts

@@ -3,11 +3,17 @@ import { Server } from '../lexicon'
 import { AppContext } from '../config'
 import algos from '../algos'
 import { validateAuth } from '../auth'
+import { AtUri } from '@atproto/uri'
 
 export default function (server: Server, ctx: AppContext) {
   server.app.bsky.feed.getFeedSkeleton(async ({ params, req }) => {
-    const algo = algos[params.feed]
-    if (!algo) {
+    const feedUri = new AtUri(params.feed)
+    const algo = algos[feedUri.rkey]
+    if (
+      feedUri.hostname !== ctx.cfg.publisherDid ||
+      feedUri.collection !== 'app.bsky.feed.generator' ||
+      !algo
+    ) {
       throw new InvalidRequestError(
         'Unsupported algorithm',
         'UnsupportedAlgorithm',

src/server.ts

@@ -63,8 +63,8 @@ export class FeedGenerator {
 
   async start(): Promise<http.Server> {
     await migrateToLatest(this.db)
-    this.firehose.run()
-    this.server = this.app.listen(this.cfg.port)
+    this.firehose.run(this.cfg.subscriptionReconnectDelay)
+    this.server = this.app.listen(this.cfg.port, this.cfg.listenhost)
     await events.once(this.server, 'listening')
     return this.server
   }

src/util/subscription.ts

@@ -54,8 +54,9 @@ export abstract class FirehoseSubscriptionBase {
 
   abstract handleEvent(evt: RepoEvent, blacksky): Promise<void>
 
-  async run() {
-    dotenv.config()
+  async run(subscriptionReconnectDelay: number) {
+    try {
+      dotenv.config()
 
     const handle = process.env.IDENTIFIER ?? ''
     const password = process.env.PASSWORD ?? ''
@@ -79,13 +80,17 @@ export abstract class FirehoseSubscriptionBase {
     for await (const evt of this.sub) {
       try {
         await this.handleEvent(evt, blacksky)
+
       } catch (err) {
         console.error('repo subscription could not handle message', err)
       }
       // update stored cursor every 20 events or so
       if (isCommit(evt) && evt.seq % 20 === 0) {
         await this.updateCursor(evt.seq)
       }
+    } catch (err) {
+      console.error('repo subscription errored', err)
+      setTimeout(() => this.run(subscriptionReconnectDelay), subscriptionReconnectDelay)
     }
   }
 

tsconfig.json

@@ -3,7 +3,7 @@
   "compilerOptions": {
     "lib": [
       "ESNext",
-    ], 
+    ],
     "outDir": "dist",
     "module": "CommonJS",
     "target": "ES6",
@@ -18,4 +18,4 @@
   "exclude": [
     "node_modules"
   ]
-}
+}