Logical separation of app view (bluesky-social#572)

* Separate app.bsky methods out into app view

* Fix websocket issue while splitting-out app view

* Start organizing app-view tables out of pds

* Move feed service, getTimeline, vote table into app-view

* Move record processing/indexing logically out of pds and into app-view

* Pull actor service functionality into app-view

* Factor app-view user functionality out of pds, replace actor w/ account service

* Tidy

packages/pds/src/api/app/bsky/actor/updateProfile.ts

@@ -30,7 +30,7 @@ export default function (server: Server, ctx: AppContext) {
           const uri = AtUri.make(did, profileNsid, 'self')
           const current = (await recordTxn.getRecord(uri, null, true))?.value
           if (current) {
-            if (!recordTxn.records.profile.matchesSchema(current)) {
+            if (!isProfile(current)) {
               // @TODO need a way to get a profile out of a broken state
               throw new InvalidRequestError('could not parse current profile')
             }
@@ -54,7 +54,7 @@ export default function (server: Server, ctx: AppContext) {
             }
           }
           updated = common.noUndefinedVals(updated)
-          if (!recordTxn.records.profile.matchesSchema(updated)) {
+          if (!isProfile(updated)) {
             throw new InvalidRequestError(
               'requested updates do not produce a valid profile doc',
             )
@@ -125,3 +125,7 @@ function unsetIfNull<T>(x: T | null | undefined, y?: T): T | undefined {
   if (x === null) return undefined
   return x ?? y
 }
+
+function isProfile(obj: unknown): obj is Profile.Record {
+  return Profile.validateRecord(obj).success
+}

packages/pds/src/api/app/bsky/graph/getMutes.ts

@@ -33,13 +33,14 @@ export default function (server: Server, ctx: AppContext) {
 
       const mutesRes = await mutesReq.execute()
 
+      // @NOTE calling into app-view, will eventually be replaced
+      const actorService = services.appView.actor(db)
+
       return {
         encoding: 'application/json',
         body: {
           cursor: keyset.packFromResult(mutesRes),
-          mutes: await services
-            .actor(db)
-            .views.actorWithInfo(mutesRes, requester),
+          mutes: await actorService.views.actorWithInfo(mutesRes, requester),
         },
       }
     },

packages/pds/src/api/app/bsky/graph/mute.ts

@@ -1,5 +1,5 @@
-import { Server } from '../../../../lexicon'
 import { InvalidRequestError } from '@atproto/xrpc-server'
+import { Server } from '../../../../lexicon'
 import AppContext from '../../../../context'
 
 export default function (server: Server, ctx: AppContext) {
@@ -10,15 +10,15 @@ export default function (server: Server, ctx: AppContext) {
       const requester = auth.credentials.did
       const { db, services } = ctx
 
-      const subject = await services.actor(db).getUser(user)
+      const subject = await services.account(db).getUser(user)
       if (!subject) {
         throw new InvalidRequestError(`Actor not found: ${user}`)
       }
       if (subject.did === requester) {
         throw new InvalidRequestError('Cannot mute oneself')
       }
 
-      await services.actor(db).mute({
+      await services.account(db).mute({
         did: subject.did,
         mutedByDid: requester,
       })

packages/pds/src/api/app/bsky/graph/unmute.ts

@@ -10,12 +10,12 @@ export default function (server: Server, ctx: AppContext) {
       const requester = auth.credentials.did
       const { db, services } = ctx
 
-      const subject = await services.actor(db).getUser(user)
+      const subject = await services.account(db).getUser(user)
       if (!subject) {
         throw new InvalidRequestError(`Actor not found: ${user}`)
       }
 
-      await services.actor(db).unmute({
+      await services.account(db).unmute({
         did: subject.did,
         mutedByDid: requester,
       })

packages/pds/src/api/app/bsky/index.ts

@@ -1,45 +1,21 @@
 import { Server } from '../../../lexicon'
 import AppContext from '../../../context'
-import getTimeline from './feed/getTimeline'
-import getAuthorFeed from './feed/getAuthorFeed'
-import getVotes from './feed/getVotes'
 import setVote from './feed/setVote'
-import getPostThread from './feed/getPostThread'
-import getProfile from './actor/getProfile'
-import getProfiles from './actor/getProfiles'
 import updateProfile from './actor/updateProfile'
-import getRepostedBy from './feed/getRepostedBy'
-import getFollowers from './graph/getFollowers'
-import getFollows from './graph/getFollows'
 import mute from './graph/mute'
 import unmute from './graph/unmute'
 import getMutes from './graph/getMutes'
-import getUsersSearch from './actor/search'
-import getUsersTypeahead from './actor/searchTypeahead'
 import getNotifications from './notification/list'
 import getNotificationCount from './notification/getCount'
-import getSuggestions from './actor/getSuggestions'
 import postNotificationsSeen from './notification/updateSeen'
 
 export default function (server: Server, ctx: AppContext) {
-  getTimeline(server, ctx)
-  getAuthorFeed(server, ctx)
-  getVotes(server, ctx)
   setVote(server, ctx)
-  getPostThread(server, ctx)
-  getProfile(server, ctx)
-  getProfiles(server, ctx)
   updateProfile(server, ctx)
-  getRepostedBy(server, ctx)
-  getFollowers(server, ctx)
-  getFollows(server, ctx)
   mute(server, ctx)
   unmute(server, ctx)
   getMutes(server, ctx)
-  getUsersSearch(server, ctx)
-  getUsersTypeahead(server, ctx)
   getNotifications(server, ctx)
   getNotificationCount(server, ctx)
-  getSuggestions(server, ctx)
   postNotificationsSeen(server, ctx)
 }

packages/pds/src/api/app/bsky/notification/list.ts

@@ -76,7 +76,8 @@ export default function (server: Server, ctx: AppContext) {
         throw new InvalidRequestError(`Could not find user: ${requester}`)
       }
 
-      const actorService = ctx.services.actor(ctx.db)
+      // @NOTE calling into app-view, will eventually be replaced
+      const actorService = ctx.services.appView.actor(ctx.db)
       const authors = await actorService.views.actorWithInfo(
         notifs.map((notif) => ({
           did: notif.authorDid,

packages/pds/src/api/app/bsky/notification/updateSeen.ts

@@ -16,7 +16,7 @@ export default function (server: Server, ctx: AppContext) {
         throw new InvalidRequestError('Invalid date')
       }
 
-      const user = await ctx.services.actor(ctx.db).getUser(requester)
+      const user = await ctx.services.account(ctx.db).getUser(requester)
       if (!user) {
         throw new InvalidRequestError(`Could not find user: ${requester}`)
       }

packages/pds/src/api/com/atproto/account/create.ts

@@ -5,7 +5,7 @@ import { Server, APP_BSKY_SYSTEM } from '../../../../lexicon'
 import { countAll } from '../../../../db/util'
 import * as lex from '../../../../lexicon/lexicons'
 import * as repo from '../../../../repo'
-import { UserAlreadyExistsError } from '../../../../services/actor'
+import { UserAlreadyExistsError } from '../../../../services/account'
 import AppContext from '../../../../context'
 
 export default function (server: Server, ctx: AppContext) {
@@ -30,7 +30,7 @@ export default function (server: Server, ctx: AppContext) {
     const now = new Date().toISOString()
 
     const result = await ctx.db.transaction(async (dbTxn) => {
-      const actorTxn = ctx.services.actor(dbTxn)
+      const actorTxn = ctx.services.account(dbTxn)
       const repoTxn = ctx.services.repo(dbTxn)
       if (ctx.cfg.inviteRequired) {
         if (!inviteCode) {

packages/pds/src/api/com/atproto/account/delete.ts

@@ -7,7 +7,7 @@ export default function (server: Server, ctx: AppContext) {
   server.com.atproto.account.delete(async ({ input }) => {
     const { did, password, token } = input.body
     const validPass = await ctx.services
-      .actor(ctx.db)
+      .account(ctx.db)
       .verifyUserPassword(did, password)
     if (!validPass) {
       throw new AuthRequiredError('Invalid did or password')
@@ -41,7 +41,7 @@ export default function (server: Server, ctx: AppContext) {
       await removeDeleteToken(dbTxn, did)
       await ctx.services.record(dbTxn).deleteForUser(did)
       await ctx.services.repo(dbTxn).deleteRepo(did)
-      await ctx.services.actor(dbTxn).deleteUser(did)
+      await ctx.services.account(dbTxn).deleteUser(did)
     })
   })
 }

packages/pds/src/api/com/atproto/account/requestDelete.ts

@@ -10,7 +10,7 @@ export default function (server: Server, ctx: AppContext) {
       const did = auth.credentials.did
       const token = getSixDigitToken()
       const requestedAt = new Date().toISOString()
-      const user = await ctx.services.actor(ctx.db).getUser(did)
+      const user = await ctx.services.account(ctx.db).getUser(did)
       if (!user) {
         throw new InvalidRequestError('user not found')
       }

packages/pds/src/api/com/atproto/account/requestPasswordReset.ts

@@ -6,7 +6,7 @@ export default function (server: Server, ctx: AppContext) {
   server.com.atproto.account.requestPasswordReset(async ({ input }) => {
     const email = input.body.email.toLowerCase()
 
-    const user = await ctx.services.actor(ctx.db).getUserByEmail(email)
+    const user = await ctx.services.account(ctx.db).getUserByEmail(email)
 
     if (user) {
       const token = getSixDigitToken()

packages/pds/src/api/com/atproto/account/resetPassword.ts

@@ -28,7 +28,7 @@ export default function (server: Server, ctx: AppContext) {
     await ctx.db.transaction(async (dbTxn) => {
       await unsetResetToken(dbTxn, tokenInfo.did)
       await ctx.services
-        .actor(dbTxn)
+        .account(dbTxn)
         .updateUserPassword(tokenInfo.did, password)
     })
   })

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

@@ -8,7 +8,7 @@ export default function (server: Server, ctx: AppContext) {
     handler: async ({ params }) => {
       const { db, services } = ctx
       const { did } = params
-      const result = await services.actor(db).getUser(did, true)
+      const result = await services.account(db).getUser(did, true)
       if (!result) throw new InvalidRequestError('Repo not found')
       return {
         encoding: 'application/json',

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

@@ -2,7 +2,7 @@ import { Server } from '../../../../lexicon'
 import AppContext from '../../../../context'
 import { SearchKeyset } from '../../../../services/util/search'
 import { sql } from 'kysely'
-import { ListKeyset } from '../../../../services/actor'
+import { ListKeyset } from '../../../../services/account'
 
 export default function (server: Server, ctx: AppContext) {
   server.com.atproto.admin.searchRepos({
@@ -14,7 +14,7 @@ export default function (server: Server, ctx: AppContext) {
 
       if (!term) {
         const results = await services
-          .actor(db)
+          .account(db)
           .list({ limit, before, includeSoftDeleted: true })
         const keyset = new ListKeyset(sql``, sql``)
 
@@ -28,7 +28,7 @@ export default function (server: Server, ctx: AppContext) {
       }
 
       const results = await services
-        .actor(db)
+        .account(db)
         .search({ term, limit, before, includeSoftDeleted: true })
       const keyset = new SearchKeyset(sql``, sql``)
 

packages/pds/src/api/com/atproto/handle.ts

@@ -3,7 +3,7 @@ import { InvalidRequestError } from '@atproto/xrpc-server'
 import * as handleLib from '@atproto/handle'
 import { Server } from '../../../lexicon'
 import AppContext from '../../../context'
-import { UserAlreadyExistsError } from '../../../services/actor'
+import { UserAlreadyExistsError } from '../../../services/account'
 import { httpLogger as log } from '../../../logger'
 import { XRPCError } from '@atproto/xrpc'
 
@@ -16,7 +16,7 @@ export default function (server: Server, ctx: AppContext) {
       // self
       did = ctx.cfg.serverDid
     } else {
-      const user = await ctx.services.actor(ctx.db).getUser(handle, true)
+      const user = await ctx.services.account(ctx.db).getUser(handle, true)
       if (user) {
         did = user.did
       } else {
@@ -80,7 +80,7 @@ export default function (server: Server, ctx: AppContext) {
 
       await ctx.db.transaction(async (dbTxn) => {
         try {
-          await ctx.services.actor(dbTxn).updateHandle(requester, handle)
+          await ctx.services.account(dbTxn).updateHandle(requester, handle)
         } catch (err) {
           if (err instanceof UserAlreadyExistsError) {
             throw new InvalidRequestError(`Handle already taken: ${handle}`)

packages/pds/src/api/com/atproto/repo.ts

@@ -1,5 +1,6 @@
 import { InvalidRequestError, AuthRequiredError } from '@atproto/xrpc-server'
 import { AtUri } from '@atproto/uri'
+import { WriteOpAction } from '@atproto/repo'
 import * as didResolver from '@atproto/did-resolver'
 import * as repo from '../../../repo'
 import { Server } from '../../../lexicon'
@@ -9,13 +10,12 @@ import {
   PreparedWrite,
 } from '../../../repo'
 import AppContext from '../../../context'
-import { WriteOpAction } from '@atproto/repo'
 
 export default function (server: Server, ctx: AppContext) {
   server.com.atproto.repo.describe(async ({ params }) => {
     const { user } = params
 
-    const userObj = await ctx.services.actor(ctx.db).getUser(user)
+    const userObj = await ctx.services.account(ctx.db).getUser(user)
     if (userObj === null) {
       throw new InvalidRequestError(`Could not find user: ${user}`)
     }
@@ -49,7 +49,7 @@ export default function (server: Server, ctx: AppContext) {
   server.com.atproto.repo.listRecords(async ({ params }) => {
     const { user, collection, limit, before, after, reverse } = params
 
-    const did = await ctx.services.actor(ctx.db).getDidForActor(user)
+    const did = await ctx.services.account(ctx.db).getDidForActor(user)
     if (!did) {
       throw new InvalidRequestError(`Could not find user: ${user}`)
     }
@@ -81,7 +81,7 @@ export default function (server: Server, ctx: AppContext) {
   server.com.atproto.repo.getRecord(async ({ params }) => {
     const { user, collection, rkey, cid } = params
 
-    const did = await ctx.services.actor(ctx.db).getDidForActor(user)
+    const did = await ctx.services.account(ctx.db).getDidForActor(user)
     if (!did) {
       throw new InvalidRequestError(`Could not find user: ${user}`)
     }

packages/pds/src/api/com/atproto/session.ts

@@ -9,7 +9,7 @@ export default function (server: Server, ctx: AppContext) {
     auth: ctx.accessVerifier,
     handler: async ({ auth }) => {
       const did = auth.credentials.did
-      const user = await ctx.services.actor(ctx.db).getUser(did)
+      const user = await ctx.services.account(ctx.db).getUser(did)
       if (!user) {
         throw new InvalidRequestError(
           `Could not find user info for account: ${did}`,
@@ -30,7 +30,7 @@ export default function (server: Server, ctx: AppContext) {
       ''
     ).toLowerCase()
     const authService = ctx.services.auth(ctx.db)
-    const actorService = ctx.services.actor(ctx.db)
+    const actorService = ctx.services.account(ctx.db)
 
     const user = identifier.includes('@')
       ? await actorService.getUserByEmail(identifier, true)
@@ -72,7 +72,7 @@ export default function (server: Server, ctx: AppContext) {
     auth: ctx.refreshVerifier,
     handler: async ({ req, auth }) => {
       const did = auth.credentials.did
-      const user = await ctx.services.actor(ctx.db).getUser(did, true)
+      const user = await ctx.services.account(ctx.db).getUser(did, true)
       if (!user) {
         throw new InvalidRequestError(
           `Could not find user info for account: ${did}`,

packages/pds/src/api/index.ts

@@ -1,13 +1,11 @@
-import { Options } from '@atproto/xrpc-server'
-import { createServer } from '../lexicon'
+import { Server } from '../lexicon'
 import comAtproto from './com/atproto'
 import appBsky from './app/bsky'
 import AppContext from '../context'
 
 export * as health from './health'
 
-export default function (ctx: AppContext, options: Options) {
-  const server = createServer(options)
+export default function (server: Server, ctx: AppContext) {
   comAtproto(server, ctx)
   appBsky(server, ctx)
   return server

packages/pds/src/api/app/bsky/actor/getProfile.ts → packages/pds/src/app-view/api/app/bsky/actor/getProfile.ts

@@ -1,7 +1,7 @@
-import { Server } from '../../../../lexicon'
 import { InvalidRequestError } from '@atproto/xrpc-server'
-import { softDeleted } from '../../../../db/util'
-import AppContext from '../../../../context'
+import { Server } from '../../../../../lexicon'
+import { softDeleted } from '../../../../../db/util'
+import AppContext from '../../../../../context'
 
 export default function (server: Server, ctx: AppContext) {
   server.app.bsky.actor.getProfile({
@@ -10,7 +10,7 @@ export default function (server: Server, ctx: AppContext) {
       const { actor } = params
       const requester = auth.credentials.did
       const { db, services } = ctx
-      const actorService = services.actor(db)
+      const actorService = services.appView.actor(db)
 
       const user = await actorService.getUser(actor, true)
 

packages/pds/src/api/app/bsky/actor/getProfiles.ts → packages/pds/src/app-view/api/app/bsky/actor/getProfiles.ts

@@ -1,5 +1,5 @@
-import { Server } from '../../../../lexicon'
-import AppContext from '../../../../context'
+import { Server } from '../../../../../lexicon'
+import AppContext from '../../../../../context'
 
 export default function (server: Server, ctx: AppContext) {
   server.app.bsky.actor.getProfiles({
@@ -8,7 +8,7 @@ export default function (server: Server, ctx: AppContext) {
       const { actors } = params
       const requester = auth.credentials.did
       const { db, services } = ctx
-      const actorService = services.actor(db)
+      const actorService = services.appView.actor(db)
 
       const users = await actorService.getUsers(actors)