@@ -1,7 +1,8 @@
-import { opendir, rmdir } from 'node:fs/promises'
+import { readdir, rmdir } from 'node:fs/promises'
 import { dirname, join } from 'node:path'
 import { describeError } from '@sim/utils/errors'
 import { isRecordLike } from '@sim/utils/object'
+import { compareStrings } from '@sim/utils/string'
 import type { OutboxEventContext } from '@/lib/core/outbox/service'
 import {
   getStorageConfig,
@@ -40,22 +41,50 @@ async function removeEmptyDirectory(path: string): Promise<void> {
   }
 }
 
-async function localKeys(prefix: string, signal: AbortSignal): Promise<string[]> {
+/** Orders keys path segment by segment, the order a sorted depth-first walk visits them in. */
+function compareKeyPaths(left: readonly string[], right: readonly string[]): number {
+  for (let index = 0; index < Math.min(left.length, right.length); index++) {
+    const order = compareStrings(left[index], right[index])
+    if (order !== 0) return order
+  }
+  return left.length - right.length
+}
+
+/**
+ * The first local keys under `prefix` in {@link compareKeyPaths} order, after `startAfter` when
+ * given, so a caller can page past objects it keeps. Prunes directories it walks to the end empty.
+ */
+async function localKeys(
+  prefix: string,
+  signal: AbortSignal,
+  startAfter?: string
+): Promise<string[]> {
+  const cursor = startAfter?.split('/')
   const keys: string[] = []
   async function visit(key: string): Promise<void> {
     signal.throwIfAborted()
-    let directory
+    let entries
     try {
-      directory = await opendir(join(UPLOAD_DIR_SERVER, key))
+      entries = await readdir(join(UPLOAD_DIR_SERVER, key), { withFileTypes: true })
     } catch (error) {
       if (describeError(error).code === 'ENOENT') return
       throw error
     }
-    for await (const entry of directory) {
+    entries.sort((left, right) => compareStrings(left.name, right.name))
+    for (const entry of entries) {
       signal.throwIfAborted()
       const child = `${key}${entry.name}`
-      if (entry.isDirectory()) await visit(`${child}/`)
-      else keys.push(child)
+      if (entry.isDirectory()) {
+        const segments = child.split('/')
+        if (
+          !cursor ||
+          segments.every((segment, index) => cursor[index] === segment) ||
+          compareKeyPaths(segments, cursor) > 0
+        )
+          await visit(`${child}/`)
+      } else if (!cursor || compareKeyPaths(child.split('/'), cursor) > 0) {
+        keys.push(child)
+      }
       if (keys.length >= STORAGE_KEY_PAGE_SIZE) return
     }
     await removeEmptyDirectory(join(UPLOAD_DIR_SERVER, key))
@@ -77,8 +106,8 @@ function keyPage(keys: string[], nextPageToken: string | undefined): StorageKeyP
 }
 
 /**
- * Lists one page of object keys under `prefix` in the bucket serving `context`. A local listing
- * has no cursor: it returns the first keys still on disk, so callers advance by deleting them.
+ * Lists one page of object keys under `prefix` in the bucket serving `context`. A local listing's
+ * page token is the page's last key, so a caller keeping objects still advances past them.
  */
 export async function listStorageKeyPage(
   prefix: string,
@@ -141,8 +170,8 @@ export async function listStorageKeyPage(
         : undefined
     )
   }
-  const keys = await localKeys(prefix, signal)
-  return { keys, done: keys.length < STORAGE_KEY_PAGE_SIZE }
+  const keys = await localKeys(prefix, signal, pageToken)
+  return keyPage(keys, keys.length < STORAGE_KEY_PAGE_SIZE ? undefined : keys[keys.length - 1])
 }
 
 /**
 
@@ -12,14 +12,10 @@ import {
   workspaceFileVersion,
   workspace as workspaceTable,
 } from '@sim/db/schema'
-import { chunkArray } from '@sim/utils/helpers'
 import { isRecordLike } from '@sim/utils/object'
-import { inArray, isNotNull, like, or, type SQL, type SQLWrapper, sql } from 'drizzle-orm'
+import { inArray, isNotNull, like, or } from 'drizzle-orm'
 import { extractStorageKey, isPublicStorageKey } from '@/lib/uploads/utils/file'
 
-const MIN_NEEDLE_LENGTH = 4
-const NEEDLES_PER_QUERY = 50
-
 /** Every string inside a URL column or a JSON settings blob, including JSON held as a string. */
 function collectStrings(value: unknown, into: string[]): void {
   if (typeof value === 'string') {
@@ -81,54 +77,40 @@ function keysNamedByUrl(value: string): string[] {
   return [...keys]
 }
 
-/**
- * The longest run of URL-safe characters after a key's root. Percent-encoding never splits such a
- * run, so any stored URL naming the key contains it; `null` when the key has none long enough.
- */
-function urlNeedle(key: string): string | null {
-  let best = ''
-  for (const run of key.slice(key.indexOf('/') + 1).match(/[A-Za-z0-9_-]+/g) ?? [])
-    if (run.length > best.length) best = run
-  return best.length >= MIN_NEEDLE_LENGTH ? best : null
-}
-
 /**
  * Keys named by the columns that hold a stored asset as a URL rather than a row:
  * `user.image`, `workspace.logo_url`, `organization.logo`, `organization.whitelabel_settings`,
  * `organization.metadata`, `custom_block.icon_url`, `chat.customizations` (deployed-chat logo
- * and avatar), and `oauth_client.icon`. With `needles`, reads only rows whose value contains one;
- * without, every row that holds a URL.
+ * and avatar), and `oauth_client.icon`. Every non-empty value is read and decoded here rather than
+ * matched in SQL, since a URL may percent-encode any character of the key it names. Only
+ * `user.image`, mostly external OAuth avatars, is narrowed, to values that point at this app's
+ * file-serve route or name a world-readable root.
  */
-async function loadUrlReferencedKeys(needles: readonly string[] | null): Promise<Set<string>> {
-  const matching = (column: SQLWrapper, fallback: SQL | undefined): SQL | undefined =>
-    needles ? or(...needles.map((needle) => sql`${column}::text LIKE ${`%${needle}%`}`)) : fallback
+async function loadUrlReferencedKeys(): Promise<Set<string>> {
   const values: string[] = []
   const sources = [
     db
       .select({ value: user.image })
       .from(user)
       .where(
-        matching(
-          user.image,
-          or(like(user.image, '%/api/files/serve/%'), like(user.image, '%profile-pictures/%'))
+        or(
+          like(user.image, '%/api/files/serve/%'),
+          like(user.image, '%profile-pictures%'),
+          like(user.image, '%og-images%'),
+          like(user.image, '%workspace-logos%'),
+          like(user.image, '%organization-logos%')
         )
       ),
     db
       .select({ value: workspaceTable.logoUrl })
       .from(workspaceTable)
-      .where(matching(workspaceTable.logoUrl, isNotNull(workspaceTable.logoUrl))),
+      .where(isNotNull(workspaceTable.logoUrl)),
     db
       .select({ value: customBlock.iconUrl })
       .from(customBlock)
-      .where(matching(customBlock.iconUrl, isNotNull(customBlock.iconUrl))),
-    db
-      .select({ value: chat.customizations })
-      .from(chat)
-      .where(matching(chat.customizations, sql`${chat.customizations}::text LIKE '%Url%'`)),
-    db
-      .select({ value: oauthClient.icon })
-      .from(oauthClient)
-      .where(matching(oauthClient.icon, isNotNull(oauthClient.icon))),
+      .where(isNotNull(customBlock.iconUrl)),
+    db.select({ value: chat.customizations }).from(chat).where(isNotNull(chat.customizations)),
+    db.select({ value: oauthClient.icon }).from(oauthClient).where(isNotNull(oauthClient.icon)),
   ]
   for (const source of sources) for (const row of await source) collectStrings(row.value, values)
   const organizations = await db
@@ -140,9 +122,9 @@ async function loadUrlReferencedKeys(needles: readonly string[] | null): Promise
     .from(organization)
     .where(
       or(
-        matching(organization.logo, isNotNull(organization.logo)),
-        matching(organization.whitelabelSettings, isNotNull(organization.whitelabelSettings)),
-        matching(organization.metadata, isNotNull(organization.metadata))
+        isNotNull(organization.logo),
+        isNotNull(organization.whitelabelSettings),
+        isNotNull(organization.metadata)
       )
     )
   for (const row of organizations) collectStrings([row.logo, row.whitelabel, row.metadata], values)
@@ -158,8 +140,8 @@ async function loadUrlReferencedKeys(needles: readonly string[] | null): Promise
  *   `workspace_file_version.key`, `document.storage_key`, `execution_large_values.key`; or
  * - it sits under a world-readable root (`isPublicStorageKey`, the same prefixes the file-serve
  *   route's `isPublicByKeyPrefix` answers without a session) and a stored asset URL names it (see
- *   {@link loadUrlReferencedKeys}). Only rows containing the key's longest URL-safe run are read,
- *   each run at most once per guard. No URL column is read for any other key:
+ *   {@link loadUrlReferencedKeys}), read at most once per guard and only once a public key comes
+ *   up. No URL column is read for any other key:
  *   `verifyFileAccess` serves it only through a row, by its key path under a workspace that still
  *   exists (which exact-key cleanup keeps; see `keysUnderLiveWorkspaces`), or by object metadata
  *   to a signed-in holder of a hand-pasted URL to a private file this deletion means to erase.
@@ -171,25 +153,7 @@ async function loadUrlReferencedKeys(needles: readonly string[] | null): Promise
  * attachments that is the intended erasure. A failed lookup throws, so nothing is deleted.
  */
 export function createStorageReferenceGuard(): (keys: readonly string[]) => Promise<Set<string>> {
-  const loadedNeedles = new Set<string>()
-  const named = new Set<string>()
-  let loadedAll = false
-  const loadNamedKeys = async (publicKeys: readonly string[]) => {
-    if (loadedAll || publicKeys.length === 0) return
-    const needles = publicKeys.map(urlNeedle)
-    if (needles.includes(null)) {
-      for (const key of await loadUrlReferencedKeys(null)) named.add(key)
-      loadedAll = true
-      return
-    }
-    const pending = [...new Set(needles.flatMap((needle) => needle ?? []))].filter(
-      (needle) => !loadedNeedles.has(needle)
-    )
-    for (const batch of chunkArray(pending, NEEDLES_PER_QUERY)) {
-      for (const key of await loadUrlReferencedKeys(batch)) named.add(key)
-      for (const needle of batch) loadedNeedles.add(needle)
-    }
-  }
+  let named: Set<string> | null = null
   return async (keys) => {
     const referenced = new Set<string>()
     if (keys.length === 0) return referenced
@@ -218,8 +182,10 @@ export function createStorageReferenceGuard(): (keys: readonly string[]) => Prom
     ]
     for (const row of rows) if (row.key) referenced.add(row.key)
     const publicKeys = list.filter(isPublicStorageKey)
-    await loadNamedKeys(publicKeys)
-    for (const key of publicKeys) if (named.has(key)) referenced.add(key)
+    if (publicKeys.length > 0) {
+      named ??= await loadUrlReferencedKeys()
+      for (const key of publicKeys) if (named.has(key)) referenced.add(key)
+    }
     return referenced
   }
 }
@@ -103,6 +103,14 @@ async function insertAdminWorkspace(id: string, ownerId: string) {
   })
 }
 
+/** A legal URL for `key` that shares no literal run with it, as a hand-written link can be. */
+function percentEncodeEveryCharacter(key: string): string {
+  return Array.from(
+    key,
+    (character) => `%${character.charCodeAt(0).toString(16).padStart(2, '0')}`
+  ).join('')
+}
+
 function fileRow(values: {
   key: string
   userId: string
@@ -112,7 +120,7 @@ function fileRow(values: {
 }) {
   return {
     id: generateId(),
-    originalName: 'fixture.txt',
+    originalName: `fixture-${generateId()}.txt`,
     contentType: 'text/plain',
     sizeBytes: 1,
     chatId: values.chatId ?? null,
@@ -154,6 +162,7 @@ async function seed() {
     legacyWorkspacelessKb: `kb/${generateId()}-legacy.txt`,
     profilePictureRowLogo: `profile-pictures/${generateId()}/org-logo.png`,
     logoNamedByUrl: `workspace-logos/${doomedId}/${generateId()}-org-logo.png`,
+    logoNamedByEncodedUrl: `workspace-logos/${doomedId}/${generateId()}-workspace-logo.png`,
   }
   await db.insert(workspaceFiles).values([
     fileRow({
@@ -217,6 +226,12 @@ async function seed() {
     .update(user)
     .set({ image: `/api/files/serve/${encodeURIComponent(keys.logoNamedByUrl)}` })
     .where(eq(user.id, ownerId))
+  await db
+    .update(workspace)
+    .set({
+      logoUrl: `/api/files/serve/s3/${percentEncodeEveryCharacter(keys.logoNamedByEncodedUrl)}`,
+    })
+    .where(eq(workspace.id, survivingId))
   for (const key of Object.values(keys)) await put(key)
   return { ownerId, departingId, doomedId, survivingId, keys }
 }
@@ -304,6 +319,7 @@ describe('account deletion storage sweep', () => {
         f.keys.legacyWorkspacelessKb,
         f.keys.profilePictureRowLogo,
         f.keys.logoNamedByUrl,
+        f.keys.logoNamedByEncodedUrl,
       ])
         expect(await readFile(join(storageRoot, key), 'utf8')).toBe('x')
       expect(
 
@@ -186,19 +186,31 @@ describe('deleted workspace storage sweep', () => {
     expect(await exists('workspace/ws1/nested/b.txt.upload-metadata.json')).toBe(false)
   })
 
-  it('spends an attempt instead of continuing when a full local page removes nothing', async () => {
-    for (let index = 0; index < 520; index++)
-      await put(`workspace/ws1/legacy/legacy file ${index}.txt`)
+  it('pages past a full page of kept objects to the ones behind it', async () => {
+    const keptKeys = Array.from(
+      { length: 520 },
+      (_, index) => `workspace/ws1/a-kept/${String(index).padStart(3, '0')}.bin`
+    )
+    const unkeptKeys = Array.from(
+      { length: 10 },
+      (_, index) => `workspace/ws1/z-unkept/${index}.bin`
+    )
+    for (const key of [...keptKeys, ...unkeptKeys]) await put(key)
+    queueTableRows(
+      schemaMock.document,
+      keptKeys.map((key) => ({ key }))
+    )
+    queueTableRows(
+      schemaMock.document,
+      keptKeys.map((key) => ({ key }))
+    )
 
-    const result = await sweep(
-      { version: 1, workspaceId: 'ws1' },
-      outboxContext({ deadlineAt: Date.now() + 120_000 })
+    await expect(sweep({ version: 1, workspaceId: 'ws1' }, outboxContext())).resolves.toBe(
+      undefined
     )
 
-    expect(result).toEqual({
-      outcome: 'deferred',
-      reason: 'Workspace storage sweep removed nothing from a full listing page',
-    })
+    for (const key of unkeptKeys) expect(await exists(key)).toBe(false)
+    for (const key of keptKeys) expect(await exists(key)).toBe(true)
   })
 
   it.each([
@@ -398,20 +410,22 @@ describe('queueing account storage cleanup', () => {
     expect(keys).toContain('copilot/abc/v1004.txt')
   })
 
-  it('only inserts inside the deletion transaction', async () => {
+  it('queues a file committed between the early read and the deletion locks, once', async () => {
+    const early = { id: 'file-1', key: 'copilot/abc/early.txt', context: 'copilot' }
+    queueTableRows(schemaMock.workspaceFiles, [early])
+    const objects = await collectAccountStorageObjects({ userId: 'user-1', workspaceIds: [] })
     queueTableRows(schemaMock.workspaceFiles, [
-      { id: 'file-1', key: 'copilot/abc/file.txt', context: 'copilot' },
+      early,
+      { id: 'file-2', key: 'copilot/abc/late.txt', context: 'copilot' },
     ])
-    const objects = await collectAccountStorageObjects({ userId: 'user-1', workspaceIds: ['ws1'] })
-    const insertOnly = { insert: dbChainMock.db.insert } as unknown as DbTransaction
 
-    await enqueueAccountStorageCleanupInTx(insertOnly, {
+    await enqueueAccountStorageCleanupInTx(dbChainMock.db as unknown as DbTransaction, {
       userId: 'user-1',
-      workspaceIds: ['ws1'],
+      workspaceIds: [],
       objects,
     })
 
-    expect(queuedObjectKeys()).toEqual(['copilot/abc/file.txt'])
+    expect(queuedObjectKeys()).toEqual(['copilot/abc/early.txt', 'copilot/abc/late.txt'])
   })
 
   it('never queues a profile picture row', async () => {