Ozone cdn invalidation

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

@@ -3,6 +3,7 @@ on:
   push:
     branches:
       - main
+      - ozone-cdn-invalidation
 env:
   REGISTRY: ${{ secrets.AWS_ECR_REGISTRY_USEAST2_PACKAGES_REGISTRY }}
   USERNAME: ${{ secrets.AWS_ECR_REGISTRY_USEAST2_PACKAGES_USERNAME }}

packages/dev-env/src/ozone.ts

@@ -62,7 +62,9 @@ export class TestOzone {
     const secrets = ozone.envToSecrets(env)
 
     // api server
-    const server = await ozone.OzoneService.create(cfg, secrets)
+    const server = await ozone.OzoneService.create(cfg, secrets, {
+      imgInvalidator: config.imgInvalidator,
+    })
     await server.start()
 
     const daemon = await ozone.OzoneDaemon.create(cfg, secrets)

packages/dev-env/src/types.ts

@@ -38,6 +38,7 @@ export type OzoneConfig = Partial<ozone.OzoneEnvironment> & {
   dbPostgresUrl: string
   migration?: string
   signingKey?: ExportableKeypair
+  imgInvalidator?: ImageInvalidator
 }
 
 export type TestServerParams = {

packages/ozone/src/config/config.ts

@@ -35,6 +35,10 @@ export const envToCfg = (env: OzoneEnvironment): OzoneConfig => {
     did: env.pdsDid,
   }
 
+  const cdnCfg: OzoneConfig['cdn'] = {
+    paths: env.cdnPaths,
+  }
+
   assert(env.didPlcUrl)
   const identityCfg: OzoneConfig['identity'] = {
     plcUrl: env.didPlcUrl,
@@ -45,6 +49,7 @@ export const envToCfg = (env: OzoneEnvironment): OzoneConfig => {
     db: dbCfg,
     appview: appviewCfg,
     pds: pdsCfg,
+    cdn: cdnCfg,
     identity: identityCfg,
   }
 }
@@ -54,6 +59,7 @@ export type OzoneConfig = {
   db: DatabaseConfig
   appview: AppviewConfig
   pds: PdsConfig | null
+  cdn: CdnConfig
   identity: IdentityConfig
 }
 
@@ -82,3 +88,7 @@ export type PdsConfig = {
 export type IdentityConfig = {
   plcUrl: string
 }
+
+export type CdnConfig = {
+  paths?: string[]
+}

packages/ozone/src/config/env.ts

@@ -1,4 +1,4 @@
-import { envInt, envStr } from '@atproto/common'
+import { envInt, envList, envStr } from '@atproto/common'
 
 export const readEnv = (): OzoneEnvironment => {
   return {
@@ -14,6 +14,7 @@ export const readEnv = (): OzoneEnvironment => {
     dbPostgresUrl: envStr('OZONE_DB_POSTGRES_URL'),
     dbPostgresSchema: envStr('OZONE_DB_POSTGRES_SCHEMA'),
     didPlcUrl: envStr('OZONE_DID_PLC_URL'),
+    cdnPaths: envList('OZONE_CDN_PATHS'),
     adminPassword: envStr('OZONE_ADMIN_PASSWORD'),
     moderatorPassword: envStr('OZONE_MODERATOR_PASSWORD'),
     triagePassword: envStr('OZONE_TRIAGE_PASSWORD'),
@@ -34,6 +35,7 @@ export type OzoneEnvironment = {
   dbPostgresUrl?: string
   dbPostgresSchema?: string
   didPlcUrl?: string
+  cdnPaths?: string[]
   adminPassword?: string
   moderatorPassword?: string
   triagePassword?: string

packages/ozone/src/context.ts

@@ -14,6 +14,7 @@ import {
   CommunicationTemplateService,
   CommunicationTemplateServiceCreator,
 } from './communication-service/template'
+import { ImageInvalidator } from './image-invalidator'
 
 export type AppContextOptions = {
   db: Database
@@ -24,6 +25,7 @@ export type AppContextOptions = {
   pdsAgent: AtpAgent | undefined
   signingKey: Keypair
   idResolver: IdResolver
+  imgInvalidator?: ImageInvalidator
   backgroundQueue: BackgroundQueue
 }
 
@@ -66,6 +68,8 @@ export class AppContext {
       appviewAgent,
       appviewAuth,
       cfg.service.did,
+      overrides?.imgInvalidator,
+      cfg.cdn.paths,
     )
 
     const communicationTemplateService = CommunicationTemplateService.creator()

packages/ozone/src/image-invalidator.ts

@@ -0,0 +1,7 @@
+// Invalidation is a general interface for propagating an image blob
+// takedown through any caches where a representation of it may be stored.
+// @NOTE this does not remove the blob from storage: just invalidates it from caches.
+// @NOTE keep in sync with same interface in aws/src/cloudfront.ts
+export interface ImageInvalidator {
+  invalidate(subject: string, paths: string[]): Promise<void>
+}

packages/ozone/src/mod-service/index.ts

@@ -41,6 +41,8 @@ import {
 import { BlobPushEvent } from '../db/schema/blob_push_event'
 import { BackgroundQueue } from '../background'
 import { EventPusher } from '../daemon'
+import { ImageInvalidator } from '../image-invalidator'
+import { httpLogger as log } from '../logger'
 
 export type ModerationServiceCreator = (db: Database) => ModerationService
 
@@ -52,6 +54,8 @@ export class ModerationService {
     public appviewAgent: AtpAgent,
     private appviewAuth: AppviewAuth,
     public serverDid: string,
+    public imgInvalidator?: ImageInvalidator,
+    public cdnPaths?: string[],
   ) {}
 
   static creator(
@@ -60,6 +64,8 @@ export class ModerationService {
     appviewAgent: AtpAgent,
     appviewAuth: AppviewAuth,
     serverDid: string,
+    imgInvalidator?: ImageInvalidator,
+    cdnPaths?: string[],
   ) {
     return (db: Database) =>
       new ModerationService(
@@ -69,6 +75,8 @@ export class ModerationService {
         appviewAgent,
         appviewAuth,
         serverDid,
+        imgInvalidator,
+        cdnPaths,
       )
   }
 
@@ -486,14 +494,38 @@ export class ModerationService {
               lastAttempted: null,
             }),
         )
-        .returning('id')
+        .returning(['id', 'subjectDid', 'subjectBlobCid', 'eventType'])
         .execute()
 
       this.db.onCommit(() => {
         this.backgroundQueue.add(async () => {
-          await Promise.all(
-            blobEvts.map((evt) => this.eventPusher.attemptBlobEvent(evt.id)),
+          await Promise.allSettled(
+            blobEvts.map((evt) =>
+              this.eventPusher
+                .attemptBlobEvent(evt.id)
+                .catch((err) =>
+                  log.error({ err, ...evt }, 'failed to push blob event'),
+                ),
+            ),
           )
+
+          if (this.imgInvalidator) {
+            await Promise.allSettled(
+              (subject.blobCids ?? []).map((cid) => {
+                const paths = (this.cdnPaths ?? []).map((path) =>
+                  path.replace('%s', subject.did).replace('%s', cid),
+                )
+                return this.imgInvalidator
+                  ?.invalidate(cid, paths)
+                  .catch((err) =>
+                    log.error(
+                      { err, paths, cid },
+                      'failed to invalidate blob on cdn',
+                    ),
+                  )
+              }),
+            )
+          }
         })
       })
     }

packages/ozone/tests/moderation.test.ts

@@ -25,6 +25,7 @@ import {
 } from '../src/lexicon/types/com/atproto/admin/defs'
 import { EventReverser } from '../src'
 import { TestOzone } from '@atproto/dev-env/src/ozone'
+import { ImageInvalidator } from '../src/image-invalidator'
 import {
   UNSPECCED_TAKEDOWN_BLOBS_LABEL,
   UNSPECCED_TAKEDOWN_LABEL,
@@ -43,6 +44,7 @@ type TakedownParams = BaseCreateReportParams &
 describe('moderation', () => {
   let network: TestNetwork
   let ozone: TestOzone
+  let mockInvalidator: MockInvalidator
   let agent: AtpAgent
   let bskyAgent: AtpAgent
   let pdsAgent: AtpAgent
@@ -155,8 +157,13 @@ describe('moderation', () => {
   }
 
   beforeAll(async () => {
+    mockInvalidator = new MockInvalidator()
     network = await TestNetwork.create({
       dbPostgresSchema: 'ozone_moderation',
+      ozone: {
+        imgInvalidator: mockInvalidator,
+        cdnPaths: ['/path1/%s/%s', '/path2/%s/%s'],
+      },
     })
     ozone = network.ozone
     agent = network.ozone.getClient()
@@ -980,6 +987,18 @@ describe('moderation', () => {
       expect(await fetchImage.json()).toEqual({ message: 'Image not found' })
     })
 
+    it('invalidates the image in the cdn', async () => {
+      const blobCid = blob.image.ref.toString()
+      expect(mockInvalidator.invalidated.length).toBe(1)
+      expect(mockInvalidator.invalidated.at(0)?.subject).toBe(blobCid)
+      expect(mockInvalidator.invalidated.at(0)?.paths.at(0)).toEqual(
+        `/path1/${sc.dids.carol}/${blobCid}`,
+      )
+      expect(mockInvalidator.invalidated.at(0)?.paths.at(1)).toEqual(
+        `/path2/${sc.dids.carol}/${blobCid}`,
+      )
+    })
+
     it('fans takedown out to pds', async () => {
       const res = await pdsAgent.api.com.atproto.admin.getSubjectStatus(
         {
@@ -1058,3 +1077,11 @@ describe('moderation', () => {
     })
   })
 })
+
+class MockInvalidator implements ImageInvalidator {
+  invalidated: { subject: string; paths: string[] }[] = []
+
+  async invalidate(subject: string, paths: string[]) {
+    this.invalidated.push({ subject, paths })
+  }
+}

services/ozone/api.js

@@ -12,6 +12,11 @@ require('dd-trace') // Only works with commonjs
 
 // Tracer code above must come before anything else
 const path = require('path')
+const {
+  BunnyInvalidator,
+  CloudfrontInvalidator,
+  MultiImageInvalidator,
+} = require('@atproto/aws')
 const {
   OzoneService,
   envToCfg,
@@ -24,7 +29,38 @@ const main = async () => {
   const env = readEnv()
   const cfg = envToCfg(env)
   const secrets = envToSecrets(env)
-  const ozone = await OzoneService.create(cfg, secrets)
+
+  // configure zero, one, or more image invalidators
+  const imgUriEndpoint = process.env.OZONE_IMG_URI_ENDPOINT
+  const bunnyAccessKey = process.env.OZONE_BUNNY_ACCESS_KEY
+  const cfDistributionId = process.env.OZONE_CF_DISTRIBUTION_ID
+
+  const imgInvalidators = []
+
+  if (bunnyAccessKey) {
+    imgInvalidators.push(
+      new BunnyInvalidator({
+        accessKey: bunnyAccessKey,
+        urlPrefix: imgUriEndpoint,
+      }),
+    )
+  }
+
+  if (cfDistributionId) {
+    imgInvalidators.push(
+      new CloudfrontInvalidator({
+        distributionId: cfDistributionId,
+        pathPrefix: imgUriEndpoint && new URL(imgUriEndpoint).pathname,
+      }),
+    )
+  }
+
+  const imgInvalidator =
+    imgInvalidators.length > 1
+      ? new MultiImageInvalidator(imgInvalidators)
+      : imgInvalidators[0]
+
+  const ozone = await OzoneService.create(cfg, secrets, { imgInvalidator })
 
   await ozone.start()
 

services/ozone/package.json

@@ -2,6 +2,7 @@
   "name": "ozone-service",
   "private": true,
   "dependencies": {
+    "@atproto/aws": "workspace:^",
     "@atproto/ozone": "workspace:^",
     "dd-trace": "3.13.2"
   }