merge main

.github/workflows/build-and-push-bsky-aws.yaml

@@ -3,6 +3,7 @@ on:
   push:
     branches:
       - main
+      - timeline-limit-1-opt
 env:
   REGISTRY: ${{ secrets.AWS_ECR_REGISTRY_USEAST2_PACKAGES_REGISTRY }}
   USERNAME: ${{ secrets.AWS_ECR_REGISTRY_USEAST2_PACKAGES_USERNAME }}

lexicons/com/atproto/admin/defs.json

@@ -558,6 +558,10 @@
         "subjectLine": {
           "type": "string",
           "description": "The subject line of the email sent to the user."
+        },
+        "comment": {
+          "type": "string",
+          "description": "Additional comment about the outgoing comm."
         }
       }
     }

lexicons/com/atproto/admin/sendEmail.json

@@ -14,7 +14,11 @@
             "recipientDid": { "type": "string", "format": "did" },
             "content": { "type": "string" },
             "subject": { "type": "string" },
-            "senderDid": { "type": "string", "format": "did" }
+            "senderDid": { "type": "string", "format": "did" },
+            "comment": {
+              "type": "string",
+              "description": "Additional comment by the sender that won't be used in the email itself but helpful to provide more context for moderators/reviewers"
+            }
           }
         }
       },

packages/api/src/client/lexicons.ts

@@ -822,6 +822,10 @@ export const schemaDict = {
             type: 'string',
             description: 'The subject line of the email sent to the user.',
           },
+          comment: {
+            type: 'string',
+            description: 'Additional comment about the outgoing comm.',
+          },
         },
       },
     },
@@ -1512,6 +1516,11 @@ export const schemaDict = {
                 type: 'string',
                 format: 'did',
               },
+              comment: {
+                type: 'string',
+                description:
+                  "Additional comment by the sender that won't be used in the email itself but helpful to provide more context for moderators/reviewers",
+              },
             },
           },
         },

packages/api/src/client/types/com/atproto/admin/defs.ts

@@ -675,6 +675,8 @@ export function validateModEventUnmute(v: unknown): ValidationResult {
 export interface ModEventEmail {
   /** The subject line of the email sent to the user. */
   subjectLine: string
+  /** Additional comment about the outgoing comm. */
+  comment?: string
   [k: string]: unknown
 }
 

packages/api/src/client/types/com/atproto/admin/sendEmail.ts

@@ -14,6 +14,8 @@ export interface InputSchema {
   content: string
   subject?: string
   senderDid: string
+  /** Additional comment by the sender that won't be used in the email itself but helpful to provide more context for moderators/reviewers */
+  comment?: string
   [k: string]: unknown
 }
 

packages/bsky/src/api/app/bsky/feed/getPostThread.ts

@@ -128,9 +128,10 @@ const composeThread = (
   // @TODO re-enable invalidReplyRoot check
   // const badReply = !!info?.invalidReplyRoot || !!info?.violatesThreadGate
   const badReply = !!info?.violatesThreadGate
-  const omitBadReply = !isAnchorPost && badReply
+  const violatesBlock = (post && blocks[post.uri]?.reply) ?? false
+  const omitBadReply = !isAnchorPost && (badReply || violatesBlock)
 
-  if (!post || blocks[post.uri]?.reply || omitBadReply) {
+  if (!post || omitBadReply) {
     return {
       $type: 'app.bsky.feed.defs#notFoundPost',
       uri: threadData.post.postUri,
@@ -156,7 +157,7 @@ const composeThread = (
   }
 
   let parent
-  if (threadData.parent && !badReply) {
+  if (threadData.parent && !badReply && !violatesBlock) {
     if (threadData.parent instanceof ParentNotFoundError) {
       parent = {
         $type: 'app.bsky.feed.defs#notFoundPost',

packages/bsky/src/api/app/bsky/feed/getTimeline.ts

@@ -1,3 +1,4 @@
+import { sql } from 'kysely'
 import { InvalidRequestError } from '@atproto/xrpc-server'
 import { Server } from '../../../../lexicon'
 import { FeedAlgorithm, FeedKeyset, getFeedDateThreshold } from '../util/feed'
@@ -55,6 +56,11 @@ export const skeleton = async (
     throw new InvalidRequestError(`Unsupported algorithm: ${algorithm}`)
   }
 
+  if (limit === 1 && !cursor) {
+    // special case for limit=1, which is often used to check if there are new items at the top of the timeline.
+    return skeletonLimit1(params, ctx)
+  }
+
   const keyset = new FeedKeyset(ref('feed_item.sortAt'), ref('feed_item.cid'))
   const sortFrom = keyset.unpack(cursor)?.primary
 
@@ -117,6 +123,54 @@ export const skeleton = async (
   }
 }
 
+// The limit=1 case is used commonly to check if there are new items at the top of the timeline.
+// Since it's so common, it's optimized here.  The most common strategy that postgres takes to
+// build a timeline is to grab all recent content from each of the user's follow, then paginate it.
+// The downside here is that it requires grabbing all recent content from all follows, even if you
+// only want a single result.  The approach here instead takes the single most recent post from
+// each of the user's follows, then sorts only those and takes the top item.
+const skeletonLimit1 = async (params: Params, ctx: Context) => {
+  const { viewer } = params
+  const { db } = ctx
+  const { ref } = db.db.dynamic
+  const creatorsQb = db.db
+    .selectFrom('follow')
+    .where('creator', '=', viewer)
+    .select('subjectDid as did')
+    .unionAll(sql`select ${viewer} as did`)
+  const feedItemsQb = db.db
+    .selectFrom(creatorsQb.as('creator'))
+    .innerJoinLateral(
+      (eb) => {
+        const keyset = new FeedKeyset(
+          ref('feed_item.sortAt'),
+          ref('feed_item.cid'),
+        )
+        const creatorFeedItemQb = eb
+          .selectFrom('feed_item')
+          .innerJoin('post', 'post.uri', 'feed_item.postUri')
+          .whereRef('feed_item.originatorDid', '=', 'creator.did')
+          .where('feed_item.sortAt', '>', getFeedDateThreshold(undefined, 2))
+          .selectAll('feed_item')
+          .select([
+            'post.replyRoot',
+            'post.replyParent',
+            'post.creator as postAuthorDid',
+          ])
+        return paginate(creatorFeedItemQb, { limit: 1, keyset }).as('result')
+      },
+      (join) => join.onTrue(),
+    )
+    .selectAll('result')
+  const keyset = new FeedKeyset(ref('result.sortAt'), ref('result.cid'))
+  const feedItems = await paginate(feedItemsQb, { limit: 1, keyset }).execute()
+  return {
+    params,
+    feedItems,
+    cursor: keyset.packFromResult(feedItems),
+  }
+}
+
 const hydration = async (
   state: SkeletonState,
   ctx: Context,

packages/bsky/src/api/health.ts

@@ -5,6 +5,20 @@ import AppContext from '../context'
 export const createRouter = (ctx: AppContext): express.Router => {
   const router = express.Router()
 
+  router.get('/', function (req, res) {
+    res.type('text/plain')
+    res.send(
+      'This is an AT Protocol Application View (AppView) for the "bsky.app" application: https://github.com/bluesky-social/atproto\n\nMost API routes are under /xrpc/',
+    )
+  })
+
+  router.get('/robots.txt', function (req, res) {
+    res.type('text/plain')
+    res.send(
+      '# Hello Friends!\n\n# Crawling the public parts of the API is allowed. HTTP 429 ("backoff") status codes are used for rate-limiting. Up to a handful concurrent requests should be ok.\nUser-agent: *\nAllow: /',
+    )
+  })
+
   router.get('/xrpc/_health', async function (req, res) {
     const { version } = ctx.cfg
     const db = ctx.db.getPrimary()

packages/bsky/src/auto-moderator/index.ts

@@ -233,7 +233,7 @@ export class AutoModerator {
     await this.pushAgent?.com.atproto.admin.emitModerationEvent({
       event: {
         $type: 'com.atproto.admin.defs#modEventLabel',
-        comment: 'automated label',
+        comment: '[AutoModerator]: Applying labels',
         createLabelVals: labels,
         negateLabelVals: [],
       },

packages/bsky/src/lexicon/lexicons.ts

@@ -822,6 +822,10 @@ export const schemaDict = {
             type: 'string',
             description: 'The subject line of the email sent to the user.',
           },
+          comment: {
+            type: 'string',
+            description: 'Additional comment about the outgoing comm.',
+          },
         },
       },
     },
@@ -1512,6 +1516,11 @@ export const schemaDict = {
                 type: 'string',
                 format: 'did',
               },
+              comment: {
+                type: 'string',
+                description:
+                  "Additional comment by the sender that won't be used in the email itself but helpful to provide more context for moderators/reviewers",
+              },
             },
           },
         },

packages/bsky/src/lexicon/types/com/atproto/admin/defs.ts

@@ -675,6 +675,8 @@ export function validateModEventUnmute(v: unknown): ValidationResult {
 export interface ModEventEmail {
   /** The subject line of the email sent to the user. */
   subjectLine: string
+  /** Additional comment about the outgoing comm. */
+  comment?: string
   [k: string]: unknown
 }
 

packages/bsky/src/lexicon/types/com/atproto/admin/sendEmail.ts

@@ -15,6 +15,8 @@ export interface InputSchema {
   content: string
   subject?: string
   senderDid: string
+  /** Additional comment by the sender that won't be used in the email itself but helpful to provide more context for moderators/reviewers */
+  comment?: string
   [k: string]: unknown
 }
 

packages/bsky/src/services/feed/views.ts

@@ -132,8 +132,8 @@ export class FeedViews {
         lists,
         viewer,
       )
-      // skip over not found & blocked posts
-      if (!post || blocks[post.uri]?.reply) {
+      // skip over not found post
+      if (!post) {
         continue
       }
       const feedPost = { post }
@@ -159,6 +159,7 @@ export class FeedViews {
       ) {
         const replyParent = this.formatMaybePostView(
           item.replyParent,
+          item.uri,
           actors,
           posts,
           threadgates,
@@ -171,6 +172,7 @@ export class FeedViews {
         )
         const replyRoot = this.formatMaybePostView(
           item.replyRoot,
+          item.uri,
           actors,
           posts,
           threadgates,
@@ -291,6 +293,7 @@ export class FeedViews {
 
   formatMaybePostView(
     uri: string,
+    replyUri: string | null,
     actors: ActorInfoMap,
     posts: PostInfoMap,
     threadgates: ThreadgateInfoMap,
@@ -320,7 +323,7 @@ export class FeedViews {
     if (
       post.author.viewer?.blockedBy ||
       post.author.viewer?.blocking ||
-      blocks[uri]?.reply
+      (replyUri !== null && blocks[replyUri]?.reply)
     ) {
       if (!opts?.usePostViewUnion) return
       return this.blockedPost(post)

packages/bsky/tests/auto-moderator/labeler.test.ts

@@ -87,7 +87,7 @@ describe('labeler', () => {
     const cid = await cidForRecord(post)
     const uri = postUri()
     autoMod.processRecord(uri, cid, post)
-    await autoMod.processAll()
+    await network.processAll()
     const labels = await getLabels(uri.toString())
     expect(labels.length).toBe(1)
     expect(labels[0]).toMatchObject({
@@ -97,6 +97,25 @@ describe('labeler', () => {
       val: 'test-label',
       neg: false,
     })
+
+    // Verify that along with applying the labels, we are also leaving trace of the label as moderation event
+    // Temporarily assign an instance of moderation service to the autoMod so that we can validate label event
+    const modSrvc = ozone.ctx.modService(ozone.ctx.db)
+    const { events } = await modSrvc.getEvents({
+      includeAllUserRecords: false,
+      subject: uri.toString(),
+      limit: 10,
+      types: [],
+    })
+    expect(events.length).toBe(1)
+    expect(events[0]).toMatchObject({
+      action: 'com.atproto.admin.defs#modEventLabel',
+      subjectUri: uri.toString(),
+      createLabelVals: 'test-label',
+      negateLabelVals: null,
+      comment: `[AutoModerator]: Applying labels`,
+      createdBy: labelerDid,
+    })
   })
 
   it('labels embeds in posts', async () => {

packages/bsky/tests/views/blocks.test.ts

@@ -112,6 +112,18 @@ describe('pds views with blocking', () => {
     expect(forSnapshot(thread)).toMatchSnapshot()
   })
 
+  it('loads blocked reply as anchor with no parent', async () => {
+    const { data: thread } = await agent.api.app.bsky.feed.getPostThread(
+      { depth: 1, uri: carolReplyToDan.ref.uriStr },
+      { headers: await network.serviceHeaders(alice) },
+    )
+    if (!isThreadViewPost(thread.thread)) {
+      throw new Error('Expected thread view post')
+    }
+    expect(thread.thread.post.uri).toEqual(carolReplyToDan.ref.uriStr)
+    expect(thread.thread.parent).toBeUndefined()
+  })
+
   it('blocks thread parent', async () => {
     // Parent is a post by dan
     const { data: thread } = await agent.api.app.bsky.feed.getPostThread(
@@ -498,7 +510,8 @@ describe('pds views with blocking', () => {
     const replyBlockedPost = timeline.feed.find(
       (item) => item.post.uri === replyBlockedUri,
     )
-    expect(replyBlockedPost).toBeUndefined()
+    assert(replyBlockedPost)
+    expect(replyBlockedPost.reply?.parent).toBeUndefined()
     const embedBlockedPost = timeline.feed.find(
       (item) => item.post.uri === embedBlockedUri,
     )

packages/bsky/tests/views/timeline.test.ts

@@ -154,6 +154,20 @@ describe('timeline views', () => {
     expect(results(paginatedAll)).toEqual(results([full.data]))
   })
 
+  it('agrees what the first item is for limit=1 and other limits', async () => {
+    const { data: timeline } = await agent.api.app.bsky.feed.getTimeline(
+      { limit: 10 },
+      { headers: await network.serviceHeaders(alice) },
+    )
+    const { data: timelineLimit1 } = await agent.api.app.bsky.feed.getTimeline(
+      { limit: 1 },
+      { headers: await network.serviceHeaders(alice) },
+    )
+    expect(timeline.feed.length).toBeGreaterThan(1)
+    expect(timelineLimit1.feed.length).toEqual(1)
+    expect(timelineLimit1.feed[0].post.uri).toBe(timeline.feed[0].post.uri)
+  })
+
   it('reflects self-labels', async () => {
     const carolTL = await agent.api.app.bsky.feed.getTimeline(
       {},

packages/pds/package.json

@@ -43,7 +43,7 @@
     "@atproto/syntax": "workspace:^",
     "@atproto/xrpc": "workspace:^",
     "@atproto/xrpc-server": "workspace:^",
-    "@did-plc/lib": "^0.0.1",
+    "@did-plc/lib": "^0.0.4",
     "better-sqlite3": "^7.6.2",
     "bytes": "^3.1.2",
     "compression": "^1.7.4",

packages/pds/src/api/com/atproto/admin/disableAccountInvites.ts

@@ -11,7 +11,7 @@ export default function (server: Server, ctx: AppContext) {
           'Account invites are managed by the entryway service',
         )
       }
-      if (!auth.credentials.admin) {
+      if (!auth.credentials.moderator) {
         throw new AuthRequiredError('Insufficient privileges')
       }
       const { account } = input.body

packages/pds/src/api/com/atproto/admin/disableInviteCodes.ts

@@ -11,7 +11,7 @@ export default function (server: Server, ctx: AppContext) {
           'Account invites are managed by the entryway service',
         )
       }
-      if (!auth.credentials.admin) {
+      if (!auth.credentials.moderator) {
         throw new AuthRequiredError('Insufficient privileges')
       }
       const { codes = [], accounts = [] } = input.body