From b3434d4e548d1efe1b11d4a157588859b6142c20 Mon Sep 17 00:00:00 2001
From: Daniel Holmgren <dtholmgren@gmail.com>
Date: Wed, 28 Feb 2024 17:26:13 -0600
Subject: [PATCH] Ozone label endpoints (#2043)

* sketching out label sequencer

* refactor sequencer

* sequencer tests

* tests

* add query labels endpoint & tests

* add pagination

* fix label formatting on temp

* tidy

* format labels

* make use listen/notify for sequencer

* fix hanging server test

* Update packages/ozone/src/api/label/queryLabels.ts

Co-authored-by: devin ivy <devinivy@gmail.com>

* pr tidy

---------

Co-authored-by: devin ivy <devinivy@gmail.com>
---
 packages/ozone/src/api/index.ts               |   4 +
 packages/ozone/src/api/label/queryLabels.ts   |  58 +++++
 .../ozone/src/api/label/subscribeLabels.ts    |  25 ++
 packages/ozone/src/api/temp/fetchLabels.ts    |   6 +-
 packages/ozone/src/config/config.ts           |   6 +
 packages/ozone/src/config/env.ts              |   6 +
 packages/ozone/src/context.ts                 |  12 +
 .../db/migrations/20231219T205730722Z-init.ts |   8 +-
 packages/ozone/src/db/schema/label.ts         |   7 +
 packages/ozone/src/index.ts                   |   2 +
 packages/ozone/src/logger.ts                  |   2 +
 packages/ozone/src/mod-service/index.ts       | 145 ++++++------
 packages/ozone/src/mod-service/util.ts        |  13 +
 packages/ozone/src/mod-service/views.ts       |   7 +-
 packages/ozone/src/sequencer/index.ts         |   2 +
 packages/ozone/src/sequencer/outbox.ts        | 122 ++++++++++
 packages/ozone/src/sequencer/sequencer.ts     | 143 +++++++++++
 packages/ozone/tests/query-labels.test.ts     | 124 ++++++++++
 packages/ozone/tests/sequencer.test.ts        | 222 ++++++++++++++++++
 packages/ozone/tests/server.test.ts           |   2 +
 20 files changed, 834 insertions(+), 82 deletions(-)
 create mode 100644 packages/ozone/src/api/label/queryLabels.ts
 create mode 100644 packages/ozone/src/api/label/subscribeLabels.ts
 create mode 100644 packages/ozone/src/mod-service/util.ts
 create mode 100644 packages/ozone/src/sequencer/index.ts
 create mode 100644 packages/ozone/src/sequencer/outbox.ts
 create mode 100644 packages/ozone/src/sequencer/sequencer.ts
 create mode 100644 packages/ozone/tests/query-labels.test.ts
 create mode 100644 packages/ozone/tests/sequencer.test.ts

diff --git a/packages/ozone/src/api/index.ts b/packages/ozone/src/api/index.ts
index 49a9d70e1fa..54e52ffe292 100644
--- a/packages/ozone/src/api/index.ts
+++ b/packages/ozone/src/api/index.ts
@@ -8,6 +8,8 @@ import getRepo from './admin/getRepo'
 import queryModerationStatuses from './admin/queryModerationStatuses'
 import queryModerationEvents from './admin/queryModerationEvents'
 import getModerationEvent from './admin/getModerationEvent'
+import queryLabels from './label/queryLabels'
+import subscribeLabels from './label/subscribeLabels'
 import fetchLabels from './temp/fetchLabels'
 import createCommunicationTemplate from './admin/createCommunicationTemplate'
 import updateCommunicationTemplate from './admin/updateCommunicationTemplate'
@@ -27,6 +29,8 @@ export default function (server: Server, ctx: AppContext) {
   getModerationEvent(server, ctx)
   queryModerationEvents(server, ctx)
   queryModerationStatuses(server, ctx)
+  queryLabels(server, ctx)
+  subscribeLabels(server, ctx)
   fetchLabels(server, ctx)
   listCommunicationTemplates(server, ctx)
   createCommunicationTemplate(server, ctx)
diff --git a/packages/ozone/src/api/label/queryLabels.ts b/packages/ozone/src/api/label/queryLabels.ts
new file mode 100644
index 00000000000..6de6380d194
--- /dev/null
+++ b/packages/ozone/src/api/label/queryLabels.ts
@@ -0,0 +1,58 @@
+import { Server } from '../../lexicon'
+import AppContext from '../../context'
+import { InvalidRequestError } from '@atproto/xrpc-server'
+import { sql } from 'kysely'
+import { formatLabel } from '../../mod-service/util'
+
+export default function (server: Server, ctx: AppContext) {
+  server.com.atproto.label.queryLabels(async ({ params }) => {
+    const { uriPatterns, sources, limit, cursor } = params
+    let builder = ctx.db.db.selectFrom('label').selectAll().limit(limit)
+    // if includes '*', then we don't need a where clause
+    if (!uriPatterns.includes('*')) {
+      builder = builder.where((qb) => {
+        // starter where clause that is always false so that we can chain `orWhere`s
+        qb = qb.where(sql`1 = 0`)
+        for (const pattern of uriPatterns) {
+          // if no '*', then we're looking for an exact match
+          if (!pattern.includes('*')) {
+            qb = qb.orWhere('uri', '=', pattern)
+          } else {
+            if (pattern.indexOf('*') < pattern.length - 1) {
+              throw new InvalidRequestError(`invalid pattern: ${pattern}`)
+            }
+            const searchPattern = pattern
+              .slice(0, -1)
+              .replaceAll('%', '') // sanitize search pattern
+              .replaceAll('_', '\\_') // escape any underscores
+            qb = qb.orWhere('uri', 'like', `${searchPattern}%`)
+          }
+        }
+        return qb
+      })
+    }
+    if (sources && sources.length > 0) {
+      builder = builder.where('src', 'in', sources)
+    }
+    if (cursor) {
+      const cursorId = parseInt(cursor, 10)
+      if (isNaN(cursorId)) {
+        throw new InvalidRequestError('invalid cursor')
+      }
+      builder = builder.where('id', '>', cursorId)
+    }
+
+    const res = await builder.execute()
+
+    const labels = res.map((l) => formatLabel(l))
+    const resCursor = res.at(-1)?.id.toString(10)
+
+    return {
+      encoding: 'application/json',
+      body: {
+        cursor: resCursor,
+        labels,
+      },
+    }
+  })
+}
diff --git a/packages/ozone/src/api/label/subscribeLabels.ts b/packages/ozone/src/api/label/subscribeLabels.ts
new file mode 100644
index 00000000000..7efb339d488
--- /dev/null
+++ b/packages/ozone/src/api/label/subscribeLabels.ts
@@ -0,0 +1,25 @@
+import { Server } from '../../lexicon'
+import AppContext from '../../context'
+import Outbox from '../../sequencer/outbox'
+import { InvalidRequestError } from '@atproto/xrpc-server'
+
+export default function (server: Server, ctx: AppContext) {
+  server.com.atproto.label.subscribeLabels(async function* ({
+    params,
+    signal,
+  }) {
+    const { cursor } = params
+    const outbox = new Outbox(ctx.sequencer)
+
+    if (cursor !== undefined) {
+      const curr = await ctx.sequencer.curr()
+      if (cursor > (curr ?? 0)) {
+        throw new InvalidRequestError('Cursor in the future.', 'FutureCursor')
+      }
+    }
+
+    for await (const evt of outbox.events(cursor, signal)) {
+      yield evt
+    }
+  })
+}
diff --git a/packages/ozone/src/api/temp/fetchLabels.ts b/packages/ozone/src/api/temp/fetchLabels.ts
index f11cb2028bb..fd0331487d1 100644
--- a/packages/ozone/src/api/temp/fetchLabels.ts
+++ b/packages/ozone/src/api/temp/fetchLabels.ts
@@ -1,5 +1,6 @@
 import { Server } from '../../lexicon'
 import AppContext from '../../context'
+import { formatLabel } from '../../mod-service/util'
 import {
   UNSPECCED_TAKEDOWN_BLOBS_LABEL,
   UNSPECCED_TAKEDOWN_LABEL,
@@ -28,10 +29,7 @@ export default function (server: Server, ctx: AppContext) {
         .limit(limit)
         .execute()
 
-      const labels = labelRes.map((l) => ({
-        ...l,
-        cid: l.cid === '' ? undefined : l.cid,
-      }))
+      const labels = labelRes.map((l) => formatLabel(l))
 
       return {
         encoding: 'application/json',
diff --git a/packages/ozone/src/config/config.ts b/packages/ozone/src/config/config.ts
index caa799b2a90..32ed8ba5cb5 100644
--- a/packages/ozone/src/config/config.ts
+++ b/packages/ozone/src/config/config.ts
@@ -19,6 +19,9 @@ export const envToCfg = (env: OzoneEnvironment): OzoneConfig => {
   const dbCfg: OzoneConfig['db'] = {
     postgresUrl: env.dbPostgresUrl,
     postgresSchema: env.dbPostgresSchema,
+    poolSize: env.dbPoolSize,
+    poolMaxUses: env.dbPoolMaxUses,
+    poolIdleTimeoutMs: env.dbPoolIdleTimeoutMs,
   }
 
   assert(env.appviewUrl)
@@ -67,6 +70,9 @@ export type ServiceConfig = {
 export type DatabaseConfig = {
   postgresUrl: string
   postgresSchema?: string
+  poolSize?: number
+  poolMaxUses?: number
+  poolIdleTimeoutMs?: number
 }
 
 export type AppviewConfig = {
diff --git a/packages/ozone/src/config/env.ts b/packages/ozone/src/config/env.ts
index 4f96ba63d53..b0ad10074eb 100644
--- a/packages/ozone/src/config/env.ts
+++ b/packages/ozone/src/config/env.ts
@@ -13,6 +13,9 @@ export const readEnv = (): OzoneEnvironment => {
     pdsDid: envStr('OZONE_PDS_DID'),
     dbPostgresUrl: envStr('OZONE_DB_POSTGRES_URL'),
     dbPostgresSchema: envStr('OZONE_DB_POSTGRES_SCHEMA'),
+    dbPoolSize: envInt('OZONE_DB_POOL_SIZE'),
+    dbPoolMaxUses: envInt('OZONE_DB_POOL_MAX_USES'),
+    dbPoolIdleTimeoutMs: envInt('OZONE_DB_POOL_IDLE_TIMEOUT_MS'),
     didPlcUrl: envStr('OZONE_DID_PLC_URL'),
     adminPassword: envStr('OZONE_ADMIN_PASSWORD'),
     moderatorPassword: envStr('OZONE_MODERATOR_PASSWORD'),
@@ -33,6 +36,9 @@ export type OzoneEnvironment = {
   pdsDid?: string
   dbPostgresUrl?: string
   dbPostgresSchema?: string
+  dbPoolSize?: number
+  dbPoolMaxUses?: number
+  dbPoolIdleTimeoutMs?: number
   didPlcUrl?: string
   adminPassword?: string
   moderatorPassword?: string
diff --git a/packages/ozone/src/context.ts b/packages/ozone/src/context.ts
index 1cb0ec1bd83..00ef4bd71ba 100644
--- a/packages/ozone/src/context.ts
+++ b/packages/ozone/src/context.ts
@@ -10,6 +10,7 @@ import * as auth from './auth'
 import { BackgroundQueue } from './background'
 import assert from 'assert'
 import { EventPusher } from './daemon'
+import Sequencer from './sequencer/sequencer'
 import {
   CommunicationTemplateService,
   CommunicationTemplateServiceCreator,
@@ -25,6 +26,7 @@ export type AppContextOptions = {
   signingKey: Keypair
   idResolver: IdResolver
   backgroundQueue: BackgroundQueue
+  sequencer: Sequencer
 }
 
 export class AppContext {
@@ -38,6 +40,9 @@ export class AppContext {
     const db = new Database({
       url: cfg.db.postgresUrl,
       schema: cfg.db.postgresSchema,
+      poolSize: cfg.db.poolSize,
+      poolMaxUses: cfg.db.poolMaxUses,
+      poolIdleTimeoutMs: cfg.db.poolIdleTimeoutMs,
     })
     const signingKey = await Secp256k1Keypair.import(secrets.signingKeyHex)
     const appviewAgent = new AtpAgent({ service: cfg.appview.url })
@@ -74,6 +79,8 @@ export class AppContext {
       plcUrl: cfg.identity.plcUrl,
     })
 
+    const sequencer = new Sequencer(db)
+
     return new AppContext(
       {
         db,
@@ -85,6 +92,7 @@ export class AppContext {
         signingKey,
         idResolver,
         backgroundQueue,
+        sequencer,
         ...(overrides ?? {}),
       },
       secrets,
@@ -139,6 +147,10 @@ export class AppContext {
     return this.opts.backgroundQueue
   }
 
+  get sequencer(): Sequencer {
+    return this.opts.sequencer
+  }
+
   get authVerifier() {
     return auth.authVerifier(this.idResolver, { aud: this.cfg.service.did })
   }
diff --git a/packages/ozone/src/db/migrations/20231219T205730722Z-init.ts b/packages/ozone/src/db/migrations/20231219T205730722Z-init.ts
index f636f40a3f4..e08c80686d2 100644
--- a/packages/ozone/src/db/migrations/20231219T205730722Z-init.ts
+++ b/packages/ozone/src/db/migrations/20231219T205730722Z-init.ts
@@ -68,13 +68,19 @@ export async function up(db: Kysely<unknown>): Promise<void> {
   // Label
   await db.schema
     .createTable('label')
+    .addColumn('id', 'bigserial', (col) => col.primaryKey())
     .addColumn('src', 'varchar', (col) => col.notNull())
     .addColumn('uri', 'varchar', (col) => col.notNull())
     .addColumn('cid', 'varchar', (col) => col.notNull())
     .addColumn('val', 'varchar', (col) => col.notNull())
     .addColumn('neg', 'boolean', (col) => col.notNull())
     .addColumn('cts', 'varchar', (col) => col.notNull())
-    .addPrimaryKeyConstraint('label_pkey', ['src', 'uri', 'cid', 'val'])
+    .execute()
+  await db.schema
+    .createIndex('unique_label_idx')
+    .unique()
+    .on('label')
+    .columns(['src', 'uri', 'cid', 'val'])
     .execute()
   await db.schema
     .createIndex('label_uri_index')
diff --git a/packages/ozone/src/db/schema/label.ts b/packages/ozone/src/db/schema/label.ts
index 0c8a398a7db..f50a6119ab3 100644
--- a/packages/ozone/src/db/schema/label.ts
+++ b/packages/ozone/src/db/schema/label.ts
@@ -1,6 +1,9 @@
+import { Generated, Selectable } from 'kysely'
+
 export const tableName = 'label'
 
 export interface Label {
+  id: Generated<number>
   src: string
   uri: string
   cid: string
@@ -9,4 +12,8 @@ export interface Label {
   cts: string
 }
 
+export type LabelRow = Selectable<Label>
+
 export type PartialDB = { [tableName]: Label }
+
+export const LabelChannel = 'label_channel' // used with notify/listen
diff --git a/packages/ozone/src/index.ts b/packages/ozone/src/index.ts
index 9ab7c13a333..cd06c62d65a 100644
--- a/packages/ozone/src/index.ts
+++ b/packages/ozone/src/index.ts
@@ -81,6 +81,7 @@ export class OzoneService {
         'background queue stats',
       )
     }, 10000)
+    await this.ctx.sequencer.start()
     const server = this.app.listen(this.ctx.cfg.service.port)
     this.server = server
     server.keepAliveTimeout = 90000
@@ -94,6 +95,7 @@ export class OzoneService {
   async destroy(): Promise<void> {
     await this.terminator?.terminate()
     await this.ctx.backgroundQueue.destroy()
+    await this.ctx.sequencer.destroy()
     await this.ctx.db.close()
     clearInterval(this.dbStatsInterval)
   }
diff --git a/packages/ozone/src/logger.ts b/packages/ozone/src/logger.ts
index 0b3e5acf52c..79c705ba5ac 100644
--- a/packages/ozone/src/logger.ts
+++ b/packages/ozone/src/logger.ts
@@ -3,6 +3,8 @@ import { subsystemLogger } from '@atproto/common'
 
 export const dbLogger: ReturnType<typeof subsystemLogger> =
   subsystemLogger('ozone:db')
+export const seqLogger: ReturnType<typeof subsystemLogger> =
+  subsystemLogger('ozone:sequencer')
 export const httpLogger: ReturnType<typeof subsystemLogger> =
   subsystemLogger('ozone')
 export const langLogger: ReturnType<typeof subsystemLogger> =
diff --git a/packages/ozone/src/mod-service/index.ts b/packages/ozone/src/mod-service/index.ts
index 7f06272d55d..0afbda79f0f 100644
--- a/packages/ozone/src/mod-service/index.ts
+++ b/packages/ozone/src/mod-service/index.ts
@@ -43,6 +43,7 @@ import { BlobPushEvent } from '../db/schema/blob_push_event'
 import { BackgroundQueue } from '../background'
 import { EventPusher } from '../daemon'
 import { jsonb } from '../db/types'
+import { LabelChannel } from '../db/schema/label'
 
 export type ModerationServiceCreator = (db: Database) => ModerationService
 
@@ -438,24 +439,22 @@ export class ModerationService {
       takedownRef,
     }))
 
-    const [repoEvts] = await Promise.all([
-      this.db.db
-        .insertInto('repo_push_event')
-        .values(values)
-        .onConflict((oc) =>
-          oc.columns(['subjectDid', 'eventType']).doUpdateSet({
-            takedownRef,
-            confirmedAt: null,
-            attempts: 0,
-            lastAttempted: null,
-          }),
-        )
-        .returning('id')
-        .execute(),
-      this.formatAndCreateLabels(subject.did, null, {
-        create: [UNSPECCED_TAKEDOWN_LABEL],
-      }),
-    ])
+    const repoEvts = await this.db.db
+      .insertInto('repo_push_event')
+      .values(values)
+      .onConflict((oc) =>
+        oc.columns(['subjectDid', 'eventType']).doUpdateSet({
+          takedownRef,
+          confirmedAt: null,
+          attempts: 0,
+          lastAttempted: null,
+        }),
+      )
+      .returning('id')
+      .execute()
+    await this.formatAndCreateLabels(subject.did, null, {
+      create: [UNSPECCED_TAKEDOWN_LABEL],
+    })
 
     this.db.onCommit(() => {
       this.backgroundQueue.add(async () => {
@@ -467,23 +466,21 @@ export class ModerationService {
   }
 
   async reverseTakedownRepo(subject: RepoSubject) {
-    const [repoEvts] = await Promise.all([
-      this.db.db
-        .updateTable('repo_push_event')
-        .where('eventType', 'in', TAKEDOWNS)
-        .where('subjectDid', '=', subject.did)
-        .set({
-          takedownRef: null,
-          confirmedAt: null,
-          attempts: 0,
-          lastAttempted: null,
-        })
-        .returning('id')
-        .execute(),
-      this.formatAndCreateLabels(subject.did, null, {
-        negate: [UNSPECCED_TAKEDOWN_LABEL],
-      }),
-    ])
+    const repoEvts = await this.db.db
+      .updateTable('repo_push_event')
+      .where('eventType', 'in', TAKEDOWNS)
+      .where('subjectDid', '=', subject.did)
+      .set({
+        takedownRef: null,
+        confirmedAt: null,
+        attempts: 0,
+        lastAttempted: null,
+      })
+      .returning('id')
+      .execute()
+    await this.formatAndCreateLabels(subject.did, null, {
+      negate: [UNSPECCED_TAKEDOWN_LABEL],
+    })
 
     this.db.onCommit(() => {
       this.backgroundQueue.add(async () => {
@@ -509,22 +506,22 @@ export class ModerationService {
     if (blobCids && blobCids.length > 0) {
       labels.push(UNSPECCED_TAKEDOWN_BLOBS_LABEL)
     }
-    const [recordEvts] = await Promise.all([
-      this.db.db
-        .insertInto('record_push_event')
-        .values(values)
-        .onConflict((oc) =>
-          oc.columns(['subjectUri', 'eventType']).doUpdateSet({
-            takedownRef,
-            confirmedAt: null,
-            attempts: 0,
-            lastAttempted: null,
-          }),
-        )
-        .returning('id')
-        .execute(),
-      this.formatAndCreateLabels(subject.uri, subject.cid, { create: labels }),
-    ])
+    const recordEvts = await this.db.db
+      .insertInto('record_push_event')
+      .values(values)
+      .onConflict((oc) =>
+        oc.columns(['subjectUri', 'eventType']).doUpdateSet({
+          takedownRef,
+          confirmedAt: null,
+          attempts: 0,
+          lastAttempted: null,
+        }),
+      )
+      .returning('id')
+      .execute()
+    await this.formatAndCreateLabels(subject.uri, subject.cid, {
+      create: labels,
+    })
 
     this.db.onCommit(() => {
       this.backgroundQueue.add(async () => {
@@ -579,29 +576,31 @@ export class ModerationService {
     if (blobCids && blobCids.length > 0) {
       labels.push(UNSPECCED_TAKEDOWN_BLOBS_LABEL)
     }
-    const [recordEvts] = await Promise.all([
-      this.db.db
-        .updateTable('record_push_event')
-        .where('eventType', 'in', TAKEDOWNS)
-        .where('subjectDid', '=', subject.did)
-        .where('subjectUri', '=', subject.uri)
-        .set({
-          takedownRef: null,
-          confirmedAt: null,
-          attempts: 0,
-          lastAttempted: null,
+    const recordEvts = await this.db.db
+      .updateTable('record_push_event')
+      .where('eventType', 'in', TAKEDOWNS)
+      .where('subjectDid', '=', subject.did)
+      .where('subjectUri', '=', subject.uri)
+      .set({
+        takedownRef: null,
+        confirmedAt: null,
+        attempts: 0,
+        lastAttempted: null,
+      })
+      .returning('id')
+      .execute()
+    await this.formatAndCreateLabels(subject.uri, subject.cid, {
+      negate: labels,
+    }),
+      this.db.onCommit(() => {
+        this.backgroundQueue.add(async () => {
+          await Promise.all(
+            recordEvts.map((evt) =>
+              this.eventPusher.attemptRecordEvent(evt.id),
+            ),
+          )
         })
-        .returning('id')
-        .execute(),
-      this.formatAndCreateLabels(subject.uri, subject.cid, { negate: labels }),
-    ])
-    this.db.onCommit(() => {
-      this.backgroundQueue.add(async () => {
-        await Promise.all(
-          recordEvts.map((evt) => this.eventPusher.attemptRecordEvent(evt.id)),
-        )
       })
-    })
 
     if (blobCids && blobCids.length > 0) {
       const blobEvts = await this.db.db
@@ -858,12 +857,14 @@ export class ModerationService {
       neg: !!l.neg,
     }))
     const { ref } = this.db.db.dynamic
+    await sql`notify ${ref(LabelChannel)}`.execute(this.db.db)
     const excluded = (col: string) => ref(`excluded.${col}`)
     await this.db.db
       .insertInto('label')
       .values(dbVals)
       .onConflict((oc) =>
         oc.columns(['src', 'uri', 'cid', 'val']).doUpdateSet({
+          id: sql`${excluded('id')}`,
           neg: sql`${excluded('neg')}`,
           cts: sql`${excluded('cts')}`,
         }),
diff --git a/packages/ozone/src/mod-service/util.ts b/packages/ozone/src/mod-service/util.ts
new file mode 100644
index 00000000000..dbb35d080eb
--- /dev/null
+++ b/packages/ozone/src/mod-service/util.ts
@@ -0,0 +1,13 @@
+import { LabelRow } from '../db/schema/label'
+import { Label } from '../lexicon/types/com/atproto/label/defs'
+
+export const formatLabel = (row: LabelRow): Label => {
+  return {
+    src: row.src,
+    uri: row.uri,
+    cid: row.cid === '' ? undefined : row.cid,
+    val: row.val,
+    neg: row.neg,
+    cts: row.cts,
+  }
+}
diff --git a/packages/ozone/src/mod-service/views.ts b/packages/ozone/src/mod-service/views.ts
index e21d2a46f5c..f1188968cbe 100644
--- a/packages/ozone/src/mod-service/views.ts
+++ b/packages/ozone/src/mod-service/views.ts
@@ -24,6 +24,7 @@ import {
 } from './types'
 import { REASONOTHER } from '../lexicon/types/com/atproto/moderation/defs'
 import { subjectFromEventRow, subjectFromStatusRow } from './subject'
+import { formatLabel } from './util'
 
 export type AppviewAuth = () => Promise<
   | {
@@ -403,11 +404,7 @@ export class ModerationViews {
       .if(!includeNeg, (qb) => qb.where('neg', '=', false))
       .selectAll()
       .execute()
-    return res.map((l) => ({
-      ...l,
-      cid: l.cid === '' ? undefined : l.cid,
-      neg: l.neg,
-    }))
+    return res.map((l) => formatLabel(l))
   }
 
   async getSubjectStatus(
diff --git a/packages/ozone/src/sequencer/index.ts b/packages/ozone/src/sequencer/index.ts
new file mode 100644
index 00000000000..0ca6e310cca
--- /dev/null
+++ b/packages/ozone/src/sequencer/index.ts
@@ -0,0 +1,2 @@
+export * from './sequencer'
+export * from './outbox'
diff --git a/packages/ozone/src/sequencer/outbox.ts b/packages/ozone/src/sequencer/outbox.ts
new file mode 100644
index 00000000000..c9ed203ca1c
--- /dev/null
+++ b/packages/ozone/src/sequencer/outbox.ts
@@ -0,0 +1,122 @@
+import { AsyncBuffer, AsyncBufferFullError } from '@atproto/common'
+import { InvalidRequestError } from '@atproto/xrpc-server'
+import { Sequencer, LabelsEvt } from './sequencer'
+
+export type OutboxOpts = {
+  maxBufferSize: number
+}
+
+export class Outbox {
+  private caughtUp = false
+  lastSeen = -1
+
+  cutoverBuffer: LabelsEvt[]
+  outBuffer: AsyncBuffer<LabelsEvt>
+
+  constructor(public sequencer: Sequencer, opts: Partial<OutboxOpts> = {}) {
+    const { maxBufferSize = 500 } = opts
+    this.cutoverBuffer = []
+    this.outBuffer = new AsyncBuffer<LabelsEvt>(maxBufferSize)
+  }
+
+  // event stream occurs in 3 phases
+  // 1. backfill events: events that have been added to the DB since the last time a connection was open.
+  // The outbox is not yet listening for new events from the sequencer
+  // 2. cutover: the outbox has caught up with where the sequencer purports to be,
+  // but the sequencer might already be halfway through sending out a round of updates.
+  // Therefore, we start accepting the sequencer's events in a buffer, while making our own request to the
+  // database to ensure we're caught up. We then dedupe the query & the buffer & stream the events in order
+  // 3. streaming: we're all caught up on historic state, so the sequencer outputs events and we
+  // immediately yield them
+  async *events(
+    backfillCursor?: number,
+    signal?: AbortSignal,
+  ): AsyncGenerator<LabelsEvt> {
+    // catch up as much as we can
+    if (backfillCursor !== undefined) {
+      for await (const evt of this.getBackfill(backfillCursor)) {
+        if (signal?.aborted) return
+        this.lastSeen = evt.seq
+        yield evt
+      }
+    } else {
+      // if not backfill, we don't need to cutover, just start streaming
+      this.caughtUp = true
+    }
+
+    // streams updates from sequencer, but buffers them for cutover as it makes a last request
+
+    const addToBuffer = (evts) => {
+      if (this.caughtUp) {
+        this.outBuffer.pushMany(evts)
+      } else {
+        this.cutoverBuffer = [...this.cutoverBuffer, ...evts]
+      }
+    }
+
+    if (!signal?.aborted) {
+      this.sequencer.on('events', addToBuffer)
+    }
+    signal?.addEventListener('abort', () =>
+      this.sequencer.off('events', addToBuffer),
+    )
+
+    const cutover = async () => {
+      // only need to perform cutover if we've been backfilling
+      if (backfillCursor !== undefined) {
+        const cutoverEvts = await this.sequencer.requestLabelRange({
+          earliestId: this.lastSeen > -1 ? this.lastSeen : backfillCursor,
+        })
+        this.outBuffer.pushMany(cutoverEvts)
+        // dont worry about dupes, we ensure order on yield
+        this.outBuffer.pushMany(this.cutoverBuffer)
+        this.caughtUp = true
+        this.cutoverBuffer = []
+      } else {
+        this.caughtUp = true
+      }
+    }
+    cutover()
+
+    while (true) {
+      try {
+        for await (const evt of this.outBuffer.events()) {
+          if (signal?.aborted) return
+          if (evt.seq > this.lastSeen) {
+            this.lastSeen = evt.seq
+            yield evt
+          }
+        }
+      } catch (err) {
+        if (err instanceof AsyncBufferFullError) {
+          throw new InvalidRequestError(
+            'Stream consumer too slow',
+            'ConsumerTooSlow',
+          )
+        } else {
+          throw err
+        }
+      }
+    }
+  }
+
+  // yields only historical events
+  async *getBackfill(backfillCursor: number) {
+    const PAGE_SIZE = 500
+    while (true) {
+      const evts = await this.sequencer.requestLabelRange({
+        earliestId: this.lastSeen > -1 ? this.lastSeen : backfillCursor,
+        limit: PAGE_SIZE,
+      })
+      for (const evt of evts) {
+        yield evt
+      }
+      // if we're within half a pagesize of the sequencer, we call it good & switch to cutover
+      const seqCursor = this.sequencer.lastSeen ?? -1
+      if (seqCursor - this.lastSeen < PAGE_SIZE / 2) break
+      if (evts.length < 1) break
+    }
+  }
+}
+
+export default Outbox
diff --git a/packages/ozone/src/sequencer/sequencer.ts b/packages/ozone/src/sequencer/sequencer.ts
new file mode 100644
index 00000000000..d7292a0d05a
--- /dev/null
+++ b/packages/ozone/src/sequencer/sequencer.ts
@@ -0,0 +1,143 @@
+import EventEmitter from 'events'
+import TypedEmitter from 'typed-emitter'
+import { seqLogger as log } from '../logger'
+import Database from '../db'
+import { Labels as LabelsEvt } from '../lexicon/types/com/atproto/label/subscribeLabels'
+import { LabelChannel, Label as LabelTable } from '../db/schema/label'
+import { Selectable } from 'kysely'
+import { formatLabel } from '../mod-service/util'
+import { PoolClient } from 'pg'
+
+export type { Labels as LabelsEvt } from '../lexicon/types/com/atproto/label/subscribeLabels'
+type LabelRow = Selectable<LabelTable>
+
+export class Sequencer extends (EventEmitter as new () => SequencerEmitter) {
+  destroyed = false
+  pollPromise: Promise<void> | undefined
+  queued = false
+  conn: PoolClient | undefined
+
+  constructor(public db: Database, public lastSeen = 0) {
+    super()
+    // note: this does not err when surpassed, just prints a warning to stderr
+    this.setMaxListeners(100)
+  }
+
+  async start() {
+    const curr = await this.curr()
+    this.lastSeen = curr ?? 0
+    this.poll()
+    this.conn = await this.db.pool.connect()
+    this.conn.query(`listen ${LabelChannel}`) // if this errors, unhandled rejection should cause process to exit
+    this.conn.on('notification', (notif) => {
+      if (notif.channel === LabelChannel) {
+        this.poll()
+      }
+    })
+  }
+
+  async destroy() {
+    if (this.destroyed) return
+    this.destroyed = true
+    if (this.conn) {
+      this.conn.release()
+      this.conn = undefined
+    }
+    if (this.pollPromise) {
+      await this.pollPromise
+    }
+    this.emit('close')
+  }
+
+  async curr(): Promise<number | null> {
+    const got = await this.db.db
+      .selectFrom('label')
+      .selectAll()
+      .orderBy('id', 'desc')
+      .limit(1)
+      .executeTakeFirst()
+    return got?.id ?? null
+  }
+
+  async next(cursor: number): Promise<LabelRow | null> {
+    const got = await this.db.db
+      .selectFrom('label')
+      .selectAll()
+      .where('id', '>', cursor)
+      .limit(1)
+      .orderBy('id', 'asc')
+      .executeTakeFirst()
+    return got || null
+  }
+
+  async requestLabelRange(opts: {
+    earliestId?: number
+    limit?: number
+  }): Promise<LabelsEvt[]> {
+    const { earliestId, limit } = opts
+
+    let seqQb = this.db.db.selectFrom('label').selectAll().orderBy('id', 'asc')
+    if (earliestId !== undefined) {
+      seqQb = seqQb.where('id', '>', earliestId)
+    }
+    if (limit !== undefined) {
+      seqQb = seqQb.limit(limit)
+    }
+
+    const rows = await seqQb.execute()
+    if (rows.length < 1) {
+      return []
+    }
+
+    const evts: LabelsEvt[] = []
+    for (const row of rows) {
+      evts.push({
+        seq: row.id,
+        labels: [formatLabel(row)],
+      })
+    }
+
+    return evts
+  }
+
+  private poll() {
+    if (this.destroyed) return
+    if (this.pollPromise) {
+      this.queued = true
+      return
+    }
+    this.queued = false
+    this.pollPromise = this.requestLabelRange({
+      earliestId: this.lastSeen,
+      limit: 500,
+    })
+      .then((evts) => {
+        this.emit('events', evts)
+        this.lastSeen = evts.at(-1)?.seq ?? this.lastSeen
+        if (evts.length > 0) {
+          this.queued = true
+        }
+      })
+      .catch((err) => {
+        log.error(
+          { err, lastSeen: this.lastSeen },
+          'sequencer failed to poll db',
+        )
+      })
+      .finally(() => {
+        this.pollPromise = undefined
+        if (this.queued) {
+          this.poll()
+        }
+      })
+  }
+}
+
+type SequencerEvents = {
+  events: (evts: LabelsEvt[]) => void
+  close: () => void
+}
+
+export type SequencerEmitter = TypedEmitter<SequencerEvents>
+
+export default Sequencer
diff --git a/packages/ozone/tests/query-labels.test.ts b/packages/ozone/tests/query-labels.test.ts
new file mode 100644
index 00000000000..a3b51d30caa
--- /dev/null
+++ b/packages/ozone/tests/query-labels.test.ts
@@ -0,0 +1,124 @@
+import AtpAgent from '@atproto/api'
+import { TestNetwork } from '@atproto/dev-env'
+import { Label } from '../src/lexicon/types/com/atproto/label/defs'
+
+describe('ozone query labels', () => {
+  let network: TestNetwork
+  let agent: AtpAgent
+
+  let labels: Label[]
+
+  beforeAll(async () => {
+    network = await TestNetwork.create({
+      dbPostgresSchema: 'ozone_query_labels',
+    })
+
+    agent = network.ozone.getClient()
+
+    labels = [
+      {
+        src: 'did:example:labeler',
+        uri: 'did:example:blah',
+        val: 'spam',
+        neg: false,
+        cts: new Date().toISOString(),
+      },
+      {
+        src: 'did:example:labeler',
+        uri: 'did:example:blah',
+        val: 'impersonation',
+        neg: false,
+        cts: new Date().toISOString(),
+      },
+      {
+        src: 'did:example:labeler',
+        uri: 'at://did:example:blah/app.bsky.feed.post/1234abcde',
+        val: 'spam',
+        neg: false,
+        cts: new Date().toISOString(),
+      },
+      {
+        src: 'did:example:labeler',
+        uri: 'at://did:example:blah/app.bsky.feed.post/1234abcfg',
+        val: 'spam',
+        neg: false,
+        cts: new Date().toISOString(),
+      },
+      {
+        src: 'did:example:labeler',
+        uri: 'at://did:example:blah/app.bsky.actor.profile/self',
+        val: 'spam',
+        neg: false,
+        cts: new Date().toISOString(),
+      },
+      {
+        src: 'did:example:labeler',
+        uri: 'did:example:thing',
+        val: 'spam',
+        neg: false,
+        cts: new Date().toISOString(),
+      },
+    ]
+
+    const modService = network.ozone.ctx.modService(network.ozone.ctx.db)
+    await modService.createLabels(labels)
+  })
+
+  afterAll(async () => {
+    await network.close()
+  })
+
+  it('returns all labels', async () => {
+    const res = await agent.api.com.atproto.label.queryLabels({
+      uriPatterns: ['*'],
+    })
+    expect(res.data.labels).toEqual(labels)
+  })
+
+  it('returns all labels even when an additional pattern is supplied', async () => {
+    const res = await agent.api.com.atproto.label.queryLabels({
+      uriPatterns: ['*', 'did:example:blah'],
+    })
+    expect(res.data.labels).toEqual(labels)
+  })
+
+  it('returns all labels that match an exact uri pattern', async () => {
+    const res = await agent.api.com.atproto.label.queryLabels({
+      uriPatterns: ['did:example:blah'],
+    })
+    expect(res.data.labels).toEqual(labels.slice(0, 2))
+  })
+
+  it('returns all labels that match one of multiple exact uris', async () => {
+    const res = await agent.api.com.atproto.label.queryLabels({
+      uriPatterns: [
+        'at://did:example:blah/app.bsky.feed.post/1234abcfg',
+        'at://did:example:blah/app.bsky.actor.profile/self',
+      ],
+    })
+    expect(res.data.labels).toEqual(labels.slice(3, 5))
+  })
+
+  it('returns all labels that match one of multiple uris, exact & glob', async () => {
+    const res = await agent.api.com.atproto.label.queryLabels({
+      uriPatterns: ['at://did:example:blah/app.bsky*', 'did:example:blah'],
+    })
+    expect(res.data.labels).toEqual(labels.slice(0, 5))
+  })
+
+  it('paginates', async () => {
+    const res1 = await agent.api.com.atproto.label.queryLabels({
+      uriPatterns: ['at://did:example:blah/app.bsky*', 'did:example:blah'],
+      limit: 3,
+    })
+    const res2 = await agent.api.com.atproto.label.queryLabels({
+      uriPatterns: ['at://did:example:blah/app.bsky*', 'did:example:blah'],
+      limit: 3,
+      cursor: res1.data.cursor,
+    })
+
+    expect([...res1.data.labels, ...res2.data.labels]).toEqual(
+      labels.slice(0, 5),
+    )
+  })
+})
diff --git a/packages/ozone/tests/sequencer.test.ts b/packages/ozone/tests/sequencer.test.ts
new file mode 100644
index 00000000000..b109aa7b7b1
--- /dev/null
+++ b/packages/ozone/tests/sequencer.test.ts
@@ -0,0 +1,222 @@
+import { TestNetwork } from '@atproto/dev-env'
+import { readFromGenerator, wait } from '@atproto/common'
+import { LabelsEvt, Sequencer } from '../src/sequencer'
+import Outbox from '../src/sequencer/outbox'
+import { randomStr } from '@atproto/crypto'
+import { Label } from '../src/lexicon/types/com/atproto/label/defs'
+
+describe('sequencer', () => {
+  let network: TestNetwork
+  let sequencer: Sequencer
+
+  let totalEvts = 0
+  let lastSeen: number
+
+  beforeAll(async () => {
+    network = await TestNetwork.create({
+      dbPostgresSchema: 'ozone_sequencer',
+    })
+    sequencer = network.ozone.ctx.sequencer
+  })
+
+  afterAll(async () => {
+    await network.close()
+  })
+
+  const loadFromDb = (lastSeen: number) => {
+    return sequencer.db.db
+      .selectFrom('label')
+      .selectAll()
+      .where('id', '>', lastSeen)
+      .orderBy('id', 'asc')
+      .execute()
+  }
+
+  const evtToDbRow = (e: LabelsEvt) => {
+    const label = e.labels[0]
+    return {
+      id: e.seq,
+      ...label,
+      cid: label.cid ? label.cid : '',
+    }
+  }
+
+  const caughtUp = (outbox: Outbox): (() => Promise<boolean>) => {
+    return async () => {
+      const lastEvt = await outbox.sequencer.curr()
+      if (lastEvt === null) return true
+      return outbox.lastSeen >= (lastEvt ?? 0)
+    }
+  }
+
+  const createLabels = async (count: number): Promise<Label[]> => {
+    const labels: Label[] = []
+    for (let i = 0; i < count; i++) {
+      const did = `did:example:${randomStr(10, 'base32')}`
+      const label = {
+        src: 'did:example:labeler',
+        uri: did,
+        val: 'spam',
+        neg: false,
+        cts: new Date().toISOString(),
+      }
+      await network.ozone.ctx.db.transaction((dbTxn) =>
+        network.ozone.ctx.modService(dbTxn).createLabels([label]),
+      )
+      labels.push(label)
+    }
+    return labels
+  }
+
+  it('sends to outbox', async () => {
+    const count = 20
+    totalEvts += count
+    await createLabels(count)
+    const outbox = new Outbox(sequencer)
+    const evts = await readFromGenerator(outbox.events(-1), caughtUp(outbox))
+    expect(evts.length).toBe(totalEvts)
+
+    const fromDb = await loadFromDb(-1)
+    expect(evts.map(evtToDbRow)).toEqual(fromDb)
+
+    lastSeen = evts.at(-1)?.seq ?? lastSeen
+  })
+
+  it('sequences negative labels', async () => {
+    const count = 5
+    totalEvts += count
+    const created = await createLabels(count)
+    const toNegate = created
+      .slice(0, 2)
+      .map((l) => ({ ...l, neg: true, cts: new Date().toISOString() }))
+    await network.ozone.ctx
+      .modService(network.ozone.ctx.db)
+      .createLabels(toNegate)
+
+    const outbox = new Outbox(sequencer)
+    const evts = await readFromGenerator(
+      outbox.events(lastSeen),
+      caughtUp(outbox),
+    )
+    expect(evts.length).toBe(count)
+
+    const fromDb = await loadFromDb(lastSeen)
+    expect(evts.map(evtToDbRow)).toEqual(fromDb)
+    expect(evts[3].labels[0].uri).toEqual(toNegate[0].uri)
+    expect(evts[3].labels[0].neg).toBe(true)
+    expect(evts[4].labels[0].uri).toEqual(toNegate[1].uri)
+    expect(evts[4].labels[0].neg).toBe(true)
+
+    lastSeen = evts.at(-1)?.seq ?? lastSeen
+  })
+
+  it('handles cut over', async () => {
+    const count = 20
+    totalEvts += count
+    const outbox = new Outbox(sequencer)
+    const createPromise = createLabels(count)
+    const [evts] = await Promise.all([
+      readFromGenerator(outbox.events(-1), caughtUp(outbox), createPromise),
+      createPromise,
+    ])
+    expect(evts.length).toBe(totalEvts)
+
+    const fromDb = await loadFromDb(-1)
+    expect(evts.map(evtToDbRow)).toEqual(fromDb)
+
+    lastSeen = evts.at(-1)?.seq ?? lastSeen
+  })
+
+  it('only gets events after cursor', async () => {
+    const count = 20
+    totalEvts += count
+    const outbox = new Outbox(sequencer)
+    const createPromise = createLabels(count)
+    const [evts] = await Promise.all([
+      readFromGenerator(
+        outbox.events(lastSeen),
+        caughtUp(outbox),
+        createPromise,
+      ),
+      createPromise,
+    ])
+
+    // +1 because we send the lastSeen date as well
+    expect(evts.length).toBe(count)
+
+    const fromDb = await loadFromDb(lastSeen)
+    expect(evts.map(evtToDbRow)).toEqual(fromDb)
+
+    lastSeen = evts.at(-1)?.seq ?? lastSeen
+  })
+
+  it('buffers events that are not being read', async () => {
+    const count = 20
+    totalEvts += count
+    const outbox = new Outbox(sequencer)
+    const createPromise = createLabels(count)
+    const gen = outbox.events(lastSeen)
+    // read enough to start streaming then wait so that the rest go into the buffer,
+    // then stream out from buffer
+    const [firstPart] = await Promise.all([
+      readFromGenerator(gen, caughtUp(outbox), createPromise, 5),
+      createPromise,
+    ])
+    const secondPart = await readFromGenerator(
+      gen,
+      caughtUp(outbox),
+      createPromise,
+    )
+    const evts = [...firstPart, ...secondPart]
+    expect(evts.length).toBe(count)
+
+    const fromDb = await loadFromDb(lastSeen)
+    expect(evts.map(evtToDbRow)).toEqual(fromDb)
+
+    lastSeen = evts.at(-1)?.seq ?? lastSeen
+  })
+
+  it('errors when buffer is overloaded', async () => {
+    const count = 20
+    totalEvts += count
+    const outbox = new Outbox(sequencer, { maxBufferSize: 5 })
+    const gen = outbox.events(lastSeen)
+    const createPromise = createLabels(count)
+    // read enough to start streaming then wait to stream rest until buffer is overloaded
+    const overloadBuffer = async () => {
+      await Promise.all([
+        readFromGenerator(gen, caughtUp(outbox), createPromise, 5),
+        createPromise,
+      ])
+      await wait(500)
+      await readFromGenerator(gen, caughtUp(outbox), createPromise)
+    }
+    await expect(overloadBuffer).rejects.toThrow('Stream consumer too slow')
+
+    await createPromise
+
+    const fromDb = await loadFromDb(lastSeen)
+    lastSeen = fromDb.at(-1)?.id ?? lastSeen
+  })
+
+  it('handles many open connections', async () => {
+    const count = 20
+    const outboxes: Outbox[] = []
+    for (let i = 0; i < 50; i++) {
+      outboxes.push(new Outbox(sequencer))
+    }
+    const createPromise = createLabels(count)
+    const readOutboxes = Promise.all(
+      outboxes.map((o) =>
+        readFromGenerator(o.events(lastSeen), caughtUp(o), createPromise),
+      ),
+    )
+    const [results] = await Promise.all([readOutboxes, createPromise])
+    const fromDb = await loadFromDb(lastSeen)
+    for (const result of results) {
+      expect(result.length).toBe(count)
+      expect(result.map(evtToDbRow)).toEqual(fromDb)
+    }
+    lastSeen = results[0].at(-1)?.seq ?? lastSeen
+  })
+})
diff --git a/packages/ozone/tests/server.test.ts b/packages/ozone/tests/server.test.ts
index 4224a636739..3b40299d4c7 100644
--- a/packages/ozone/tests/server.test.ts
+++ b/packages/ozone/tests/server.test.ts
@@ -55,6 +55,8 @@ describe('server', () => {
   })
 
   it('healthcheck fails when database is unavailable.', async () => {
+    // destory sequencer to release connection that would prevent the db from closing
+    await ozone.ctx.sequencer.destroy()
     await ozone.ctx.db.close()
     let error: AxiosError
     try {