diff --git a/src/state/session/index.tsx b/src/state/session/index.tsx index 00879ea0b3..ee3e480c56 100644 --- a/src/state/session/index.tsx +++ b/src/state/session/index.tsx @@ -106,11 +106,13 @@ class SessionStore { this.repository.getSnapshot(), nextSnapshot, ) - if (result.status === 'pending') { - this.storageErrorListeners.forEach(listener => - listener(result.error.kind), - ) - } + void Promise.resolve(result).then(commitResult => { + if (commitResult.status === 'pending') { + this.storageErrorListeners.forEach(listener => + listener(commitResult.error.kind), + ) + } + }) this.listeners.forEach(listener => listener()) } diff --git a/src/state/session/storage/__tests__/repository-test.ts b/src/state/session/storage/__tests__/repository-test.ts index 37a9bc0be7..2186bddb02 100644 --- a/src/state/session/storage/__tests__/repository-test.ts +++ b/src/state/session/storage/__tests__/repository-test.ts @@ -275,4 +275,69 @@ describe('NativeSessionRepository', () => { dids: [], }) }) + + it('cancels an older pending credential write when clear fails', async () => { + const repository = new NativeSessionRepository() + const active = {accounts: [alice], currentDid: alice.did} + await repository.open(active) + const keys = accountKeys(alice.did) + + mockFailKey = keys.access + expect( + repository.commit(active, { + accounts: [{...alice, accessJwt: 'new-access'}], + currentDid: alice.did, + }).status, + ).toBe('pending') + + mockFailKey = SESSION_INDEX_KEY + await expect(repository.clear()).rejects.toThrow('disk full') + expect(repository.getSnapshot()).toEqual({ + accounts: [], + currentDid: undefined, + }) + + expect(repository.commit(active, active).status).toBe('pending') + expect(repository.getSnapshot()).toEqual({ + accounts: [], + currentDid: undefined, + }) + + mockFailKey = undefined + expect(repository.retryPending()).toEqual({status: 'committed'}) + expect(mockValues.get(keys.refresh)).toBe('') + expect(mockValues.get(keys.access)).toBe('') + expect(mockValues.get(keys.descriptor)).toBe('') + expect(JSON.parse(mockValues.get(SESSION_INDEX_KEY)!)).toEqual({ + version: 1, + dids: [], + }) + }) + + it('tombstones credentials from a superseded partial account write', async () => { + const repository = new NativeSessionRepository() + await repository.open() + const keys = accountKeys(alice.did) + + mockFailKey = SESSION_INDEX_KEY + expect( + repository.commit( + {accounts: []}, + {accounts: [alice], currentDid: alice.did}, + ).status, + ).toBe('pending') + expect(mockValues.get(keys.refresh)).toBe(alice.refreshJwt) + expect(mockValues.get(keys.access)).toBe(alice.accessJwt) + + mockFailKey = undefined + expect( + repository.commit( + {accounts: [alice], currentDid: alice.did}, + {accounts: [], currentDid: undefined}, + ), + ).toEqual({status: 'committed'}) + expect(mockValues.get(keys.refresh)).toBe('') + expect(mockValues.get(keys.access)).toBe('') + expect(mockValues.get(keys.descriptor)).toBe('') + }) }) diff --git a/src/state/session/storage/__tests__/repository-web-test.ts b/src/state/session/storage/__tests__/repository-web-test.ts new file mode 100644 index 0000000000..868491bae0 --- /dev/null +++ b/src/state/session/storage/__tests__/repository-web-test.ts @@ -0,0 +1,344 @@ +import {afterEach, beforeEach, describe, expect, it, jest} from '@jest/globals' + +import {WebSessionRepository} from '../repository.web' +import {type SessionAccount, type SessionSnapshot} from '../schema' + +const STORAGE_KEY = 'BSKY_SESSION_STORAGE_V1' +const values = new Map() +let failRead = false +let failWrite = false +let lockQueue = Promise.resolve(undefined) + +const storageLocks = { + request: jest.fn((_name: string, callback: () => unknown) => { + const result = lockQueue.then(callback) + lockQueue = result.then( + () => undefined, + () => undefined, + ) + return result + }), +} + +const storage = { + getItem: jest.fn((key: string) => { + if (failRead) throw new Error('localStorage unavailable') + return values.get(key) ?? null + }), + setItem: jest.fn((key: string, value: string) => { + if (failWrite) throw new Error('localStorage unavailable') + values.set(key, value) + }), +} + +const alice: SessionAccount = { + service: 'https://bsky.social', + did: 'did:plc:alice', + handle: 'alice.test', + refreshJwt: 'alice-refresh', + accessJwt: 'alice-access', + emailConfirmed: false, +} +const bob: SessionAccount = { + service: 'https://bsky.social', + did: 'did:plc:bob', + handle: 'bob.test', + refreshJwt: 'bob-refresh', + accessJwt: 'bob-access', +} +const charlie: SessionAccount = { + service: 'https://bsky.social', + did: 'did:plc:charlie', + handle: 'charlie.test', + refreshJwt: 'charlie-refresh', + accessJwt: 'charlie-access', +} + +beforeEach(() => { + jest.useFakeTimers() + values.clear() + failRead = false + failWrite = false + lockQueue = Promise.resolve(undefined) + storageLocks.request.mockClear() + storage.getItem.mockClear() + storage.setItem.mockClear() + Object.defineProperty(globalThis, 'localStorage', { + configurable: true, + value: storage, + }) + Object.defineProperty(globalThis.navigator, 'locks', { + configurable: true, + value: storageLocks, + }) + if (!globalThis.window) { + Object.defineProperty(globalThis, 'window', { + configurable: true, + value: {}, + }) + } + Object.defineProperty(globalThis.window, 'addEventListener', { + configurable: true, + value: jest.fn(), + }) +}) + +afterEach(() => { + jest.clearAllTimers() + jest.useRealTimers() +}) + +describe('WebSessionRepository', () => { + it('repairs corrupt storage and starts logged out', async () => { + values.set(STORAGE_KEY, '{invalid json') + const repository = new WebSessionRepository() + + await expect( + repository.open({accounts: [alice], currentDid: alice.did}), + ).resolves.toEqual({ + status: 'ready', + snapshot: {accounts: [], currentDid: undefined}, + shouldScrubLegacy: true, + }) + expect(readStoredSnapshot()).toEqual({accounts: []}) + }) + + it('keeps a legacy session in memory when localStorage is unavailable', async () => { + failRead = true + failWrite = true + const repository = new WebSessionRepository() + const legacy = {accounts: [alice], currentDid: alice.did} + + await expect(repository.open(legacy)).resolves.toEqual({ + status: 'ready', + snapshot: legacy, + shouldScrubLegacy: false, + }) + expect(repository.getSnapshot()).toEqual(legacy) + }) + + it('rebases local changes onto a newer snapshot from another tab', async () => { + const previous: SessionSnapshot = { + accounts: [alice, bob], + currentDid: alice.did, + } + const repository = new WebSessionRepository() + await repository.open(previous) + + const latest: SessionSnapshot = { + accounts: [ + {...alice, refreshJwt: 'fresh-refresh', accessJwt: 'fresh-access'}, + bob, + charlie, + ], + currentDid: charlie.did, + } + values.set(STORAGE_KEY, JSON.stringify(latest)) + + const next: SessionSnapshot = { + accounts: [{...alice, emailConfirmed: true}], + currentDid: alice.did, + } + await expect(repository.commit(previous, next)).resolves.toEqual({ + status: 'committed', + }) + + expect(readStoredSnapshot()).toEqual({ + accounts: [ + { + ...alice, + refreshJwt: 'fresh-refresh', + accessJwt: 'fresh-access', + emailConfirmed: true, + }, + charlie, + ], + currentDid: charlie.did, + }) + }) + + it('serializes concurrent commits from multiple tabs', async () => { + const previous: SessionSnapshot = { + accounts: [alice, bob], + currentDid: alice.did, + } + const firstTab = new WebSessionRepository() + const secondTab = new WebSessionRepository() + await firstTab.open(previous) + await secondTab.open(previous) + + await Promise.all([ + firstTab.commit(previous, { + accounts: [{...alice, emailConfirmed: true}, bob], + currentDid: alice.did, + }), + secondTab.commit(previous, { + accounts: [alice, {...bob, handle: 'new-bob.test'}], + currentDid: alice.did, + }), + ]) + + expect(readStoredSnapshot()).toEqual({ + accounts: [ + {...alice, emailConfirmed: true}, + {...bob, handle: 'new-bob.test'}, + ], + currentDid: alice.did, + }) + expect(storageLocks.request).toHaveBeenCalledWith( + 'bsky-session:BSKY_SESSION_STORAGE_V1', + expect.any(Function), + ) + }) + + it('does not restore credentials revoked by another tab', async () => { + const previous: SessionSnapshot = { + accounts: [alice], + currentDid: alice.did, + } + const repository = new WebSessionRepository() + await repository.open(previous) + + values.set( + STORAGE_KEY, + JSON.stringify({ + accounts: [{...alice, refreshJwt: undefined, accessJwt: undefined}], + currentDid: undefined, + }), + ) + await repository.commit(previous, { + accounts: [ + {...alice, refreshJwt: 'fresh-refresh', accessJwt: 'fresh-access'}, + ], + currentDid: alice.did, + }) + + const { + refreshJwt: _refreshJwt, + accessJwt: _accessJwt, + ...loggedOutAlice + } = alice + expect(readStoredSnapshot()).toEqual({ + accounts: [loggedOutAlice], + }) + }) + + it('preserves credential revocation during concurrent account adds', async () => { + const repository = new WebSessionRepository() + const previous: SessionSnapshot = {accounts: [], currentDid: undefined} + await repository.open(previous) + + const { + refreshJwt: _refreshJwt, + accessJwt: _accessJwt, + ...loggedOutAlice + } = alice + values.set( + STORAGE_KEY, + JSON.stringify({accounts: [loggedOutAlice], currentDid: undefined}), + ) + + await repository.commit(previous, { + accounts: [alice], + currentDid: alice.did, + }) + expect(readStoredSnapshot()).toEqual({ + accounts: [loggedOutAlice], + currentDid: alice.did, + }) + }) + + it('does not restore an account deleted by another tab', async () => { + const previous: SessionSnapshot = { + accounts: [alice, bob], + currentDid: alice.did, + } + const repository = new WebSessionRepository() + await repository.open(previous) + + values.set( + STORAGE_KEY, + JSON.stringify({accounts: [bob], currentDid: bob.did}), + ) + const editedAlice = {...alice, emailConfirmed: true} + const next: SessionSnapshot = { + accounts: [editedAlice, bob], + currentDid: alice.did, + } + + await expect(repository.commit(previous, next)).resolves.toEqual({ + status: 'committed', + }) + expect(readStoredSnapshot()).toEqual({ + accounts: [bob], + currentDid: bob.did, + }) + }) + + it('replaces an older pending commit with a failed clear', async () => { + const previous: SessionSnapshot = { + accounts: [alice], + currentDid: alice.did, + } + const repository = new WebSessionRepository() + await repository.open(previous) + + failWrite = true + await expect( + repository.commit(previous, { + accounts: [{...alice, accessJwt: 'new-access'}], + currentDid: alice.did, + }), + ).resolves.toMatchObject({status: 'pending'}) + + await expect(repository.clear()).rejects.toThrow( + 'session storage clear failed: write-failed', + ) + expect(repository.getSnapshot()).toEqual({ + accounts: [], + currentDid: undefined, + }) + + await expect(repository.commit(previous, previous)).resolves.toMatchObject({ + status: 'pending', + }) + expect(repository.getSnapshot()).toEqual({ + accounts: [], + currentDid: undefined, + }) + + failWrite = false + await expect(repository.retryPending()).resolves.toEqual({ + status: 'committed', + }) + expect(readStoredSnapshot()).toEqual({accounts: []}) + expect(repository.getSnapshot()).toEqual({accounts: []}) + }) + + it('does not overwrite newer storage when initial persistence retries', async () => { + failWrite = true + const repository = new WebSessionRepository() + const legacy = {accounts: [alice], currentDid: alice.did} + const onLegacyMigrationComplete = jest.fn() + await repository.open(legacy, onLegacyMigrationComplete) + expect(onLegacyMigrationComplete).not.toHaveBeenCalled() + + const refreshed = { + accounts: [{...alice, accessJwt: 'fresh-access'}], + currentDid: alice.did, + } + failWrite = false + values.set(STORAGE_KEY, JSON.stringify(refreshed)) + + await expect(repository.retryPending()).resolves.toEqual({ + status: 'committed', + }) + expect(readStoredSnapshot()).toEqual(refreshed) + expect(repository.getSnapshot()).toEqual(refreshed) + expect(onLegacyMigrationComplete).toHaveBeenCalledTimes(1) + }) +}) + +function readStoredSnapshot() { + return JSON.parse(values.get(STORAGE_KEY)!) +} diff --git a/src/state/session/storage/index.ts b/src/state/session/storage/index.ts index ff42885945..53070d9d7d 100644 --- a/src/state/session/storage/index.ts +++ b/src/state/session/storage/index.ts @@ -18,7 +18,14 @@ export async function initSessionRepository() { ? legacyCurrentDid : undefined, } - const result = await repository.open(legacySnapshot) + const scrubLegacy = () => + persisted.write('session', { + accounts: [], + currentAccount: undefined, + }) + const result = await repository.open(legacySnapshot, () => { + void scrubLegacy() + }) if (result.status === 'unavailable') { throw new Error(`session storage unavailable: ${result.error.kind}`) } @@ -26,10 +33,7 @@ export async function initSessionRepository() { if (result.shouldScrubLegacy) { // The new repository has been read back successfully. Scrub the old blob // so future preference writes cannot keep rewriting bearer credentials. - await persisted.write('session', { - accounts: [], - currentAccount: undefined, - }) + await scrubLegacy() } initialized = true return repository diff --git a/src/state/session/storage/repository.ts b/src/state/session/storage/repository.ts index 1d7d6af383..df10ddf9c7 100644 --- a/src/state/session/storage/repository.ts +++ b/src/state/session/storage/repository.ts @@ -39,12 +39,17 @@ export class NativeSessionRepository implements SessionRepository { private persistedSnapshot: SessionSnapshot = EMPTY_SNAPSHOT private hasPersistedIndex = false private pendingSnapshot: SessionSnapshot | undefined + private pendingClearDids: string[] | undefined + private possiblyWrittenDids = new Set() private retryTimer: ReturnType | undefined private listeners = new Set<(snapshot: SessionSnapshot) => void>() constructor() { onAppStateChange(state => { - if (state === 'active' && this.pendingSnapshot) { + if ( + state === 'active' && + (this.pendingSnapshot || this.pendingClearDids) + ) { this.retryPending() } }) @@ -101,12 +106,18 @@ export class NativeSessionRepository implements SessionRepository { _previous: SessionSnapshot, next: SessionSnapshot, ): SessionStorageCommitResult { + if (this.pendingClearDids) { + this.snapshot = EMPTY_SNAPSHOT + return this.retryPending() + } this.snapshot = next + this.trackPossiblyWrittenAccounts(next) try { this.writeSnapshot(this.persistedSnapshot, next) this.persistedSnapshot = next this.hasPersistedIndex = true this.pendingSnapshot = undefined + this.possiblyWrittenDids.clear() this.cancelRetry() return {status: 'committed'} } catch (cause) { @@ -119,15 +130,24 @@ export class NativeSessionRepository implements SessionRepository { } retryPending(): SessionStorageCommitResult { - if (!this.pendingSnapshot) { + if (!this.pendingSnapshot && !this.pendingClearDids) { return {status: 'committed'} } - const next = this.pendingSnapshot try { - this.writeSnapshot(this.persistedSnapshot, next) - this.persistedSnapshot = next + if (this.pendingClearDids) { + this.writeClear(this.pendingClearDids) + this.snapshot = EMPTY_SNAPSHOT + this.persistedSnapshot = EMPTY_SNAPSHOT + this.pendingClearDids = undefined + } else { + const next = this.pendingSnapshot! + this.trackPossiblyWrittenAccounts(next) + this.writeSnapshot(this.persistedSnapshot, next) + this.persistedSnapshot = next + } this.hasPersistedIndex = true this.pendingSnapshot = undefined + this.possiblyWrittenDids.clear() this.cancelRetry() return {status: 'committed'} } catch (cause) { @@ -151,26 +171,24 @@ export class NativeSessionRepository implements SessionRepository { ...new Set([ ...this.persistedSnapshot.accounts.map(account => account.did), ...this.snapshot.accounts.map(account => account.did), + ...this.possiblyWrittenDids, ]), ] + this.snapshot = EMPTY_SNAPSHOT + this.pendingSnapshot = undefined + this.pendingClearDids = dids + this.cancelRetry() try { - SecureStore.setItem( - SESSION_INDEX_KEY, - JSON.stringify(toStoredIndex(EMPTY_SNAPSHOT, dids)), - ) - dids.forEach(tombstoneAccount) - SecureStore.setItem( - SESSION_INDEX_KEY, - JSON.stringify(toStoredIndex(EMPTY_SNAPSHOT)), - ) - this.snapshot = EMPTY_SNAPSHOT + this.writeClear(dids) this.persistedSnapshot = EMPTY_SNAPSHOT this.hasPersistedIndex = true - this.pendingSnapshot = undefined + this.pendingClearDids = undefined + this.possiblyWrittenDids.clear() this.cancelRetry() } catch (cause) { const error = storageError('clear', cause) logStorageError(error) + this.scheduleRetry() throw cause } } @@ -233,9 +251,14 @@ export class NativeSessionRepository implements SessionRepository { private writeSnapshot(previous: SessionSnapshot, next: SessionSnapshot) { const previousByDid = new Map(previous.accounts.map(a => [a.did, a])) const nextDids = new Set(next.accounts.map(a => a.did)) - const retiredDids = previous.accounts - .filter(account => !nextDids.has(account.did)) - .map(account => account.did) + const retiredDids = [ + ...new Set([ + ...previous.accounts + .filter(account => !nextDids.has(account.did)) + .map(account => account.did), + ...[...this.possiblyWrittenDids].filter(did => !nextDids.has(did)), + ]), + ] const revokedDids = next.accounts .filter(account => { const prior = previousByDid.get(account.did) @@ -274,6 +297,7 @@ export class NativeSessionRepository implements SessionRepository { if ( !this.hasPersistedIndex || + retiredDids.length > 0 || JSON.stringify(previous) !== JSON.stringify(next) ) { // Publishing the index is the commit point. `retiredDids` makes token @@ -293,6 +317,29 @@ export class NativeSessionRepository implements SessionRepository { } } + private writeClear(dids: string[]) { + SecureStore.setItem( + SESSION_INDEX_KEY, + JSON.stringify(toStoredIndex(EMPTY_SNAPSHOT, dids)), + ) + dids.forEach(tombstoneAccount) + SecureStore.setItem( + SESSION_INDEX_KEY, + JSON.stringify(toStoredIndex(EMPTY_SNAPSHOT)), + ) + } + + private trackPossiblyWrittenAccounts(next: SessionSnapshot) { + const persistedDids = new Set( + this.persistedSnapshot.accounts.map(account => account.did), + ) + for (const account of next.accounts) { + if (!persistedDids.has(account.did)) { + this.possiblyWrittenDids.add(account.did) + } + } + } + private scheduleRetry() { if (this.retryTimer) return this.retryTimer = setTimeout(() => { diff --git a/src/state/session/storage/repository.web.ts b/src/state/session/storage/repository.web.ts index 3c11147689..cdc8310f6e 100644 --- a/src/state/session/storage/repository.web.ts +++ b/src/state/session/storage/repository.web.ts @@ -1,6 +1,13 @@ +import {createStore, update} from 'idb-keyval' + import BroadcastChannel from '#/lib/broadcast' import {logger} from '#/logger' -import {type SessionSnapshot, sessionSnapshotSchema} from './schema' +import { + type SessionAccount, + sessionAccountSchema, + type SessionSnapshot, + sessionSnapshotSchema, +} from './schema' import { type SessionRepository, type SessionStorageCommitResult, @@ -9,86 +16,94 @@ import { } from './types' const STORAGE_KEY = 'BSKY_SESSION_STORAGE_V1' +const STORAGE_LOCK_NAME = `bsky-session:${STORAGE_KEY}` +const fallbackLockStore = createStore('BSKY_SESSION_STORAGE_LOCKS', 'locks') const CHANNEL_NAME = 'BSKY_SESSION_BROADCAST_CHANNEL' const UPDATE_EVENT = 'session-update-v1' const EMPTY_SNAPSHOT: SessionSnapshot = {accounts: [], currentDid: undefined} const RETRY_DELAY = 5_000 +type PendingCommit = { + previous: SessionSnapshot + next: SessionSnapshot + replace?: boolean +} + export class WebSessionRepository implements SessionRepository { private snapshot: SessionSnapshot = EMPTY_SNAPSHOT - private pendingSnapshot: SessionSnapshot | undefined + private pendingCommit: PendingCommit | undefined + private activeOperations = new Map< + PendingCommit, + Promise + >() private listeners = new Set<(snapshot: SessionSnapshot) => void>() private retryTimer: ReturnType | undefined private opened = false + private legacyMigrationPending = false + private onLegacyMigrationComplete: (() => void) | undefined private broadcast = new BroadcastChannel(CHANNEL_NAME) - // async to keep one repository contract across native and web. - // eslint-disable-next-line @typescript-eslint/require-await - async open(legacy?: SessionSnapshot): Promise { + async open( + legacy?: SessionSnapshot, + onLegacyMigrationComplete?: () => void, + ): Promise { + this.onLegacyMigrationComplete = onLegacyMigrationComplete + let shouldScrubLegacy = false try { const stored = readFromStorage() if (stored) { this.snapshot = stored + shouldScrubLegacy = Boolean(legacy?.accounts.length) } else { this.snapshot = legacy ?? EMPTY_SNAPSHOT - writeToStorage(this.snapshot) - } - if (!this.opened) { - this.opened = true - this.broadcast.onmessage = this.onBroadcastMessage - window.addEventListener('storage', this.onStorage) - } - return { - status: 'ready', - snapshot: this.snapshot, - shouldScrubLegacy: Boolean(legacy?.accounts.length), + shouldScrubLegacy = + (await this.persistInitialSnapshot(this.snapshot)) && + Boolean(legacy?.accounts.length) + this.legacyMigrationPending = + !shouldScrubLegacy && Boolean(legacy?.accounts.length) } } catch (cause) { - const error = storageError('open', cause) - logStorageError(error) - return {status: 'unavailable', error} + logStorageError(storageError('open', cause)) + // A corrupt dedicated session key proves migration already committed; + // do not resurrect legacy credentials. If localStorage itself is + // unavailable, keep the legacy snapshot alive in memory for this tab. + this.snapshot = + cause instanceof InvalidWebSessionStorageDataError + ? EMPTY_SNAPSHOT + : (legacy ?? EMPTY_SNAPSHOT) + shouldScrubLegacy = + (await this.persistInitialSnapshot(this.snapshot)) && + Boolean(legacy?.accounts.length) + this.legacyMigrationPending = + !shouldScrubLegacy && Boolean(legacy?.accounts.length) } + this.attachListeners() + return {status: 'ready', snapshot: this.snapshot, shouldScrubLegacy} } getSnapshot(): SessionSnapshot { return this.snapshot } - commit( - _previous: SessionSnapshot, + async commit( + previous: SessionSnapshot, next: SessionSnapshot, - ): SessionStorageCommitResult { - this.snapshot = next - try { - writeToStorage(next) - this.pendingSnapshot = undefined - this.cancelRetry() - this.broadcast.postMessage({event: UPDATE_EVENT}) - return {status: 'committed'} - } catch (cause) { - this.pendingSnapshot = next - const error = storageError('commit', cause) - logStorageError(error) - this.scheduleRetry() - return {status: 'pending', error} + ): Promise { + if (this.pendingCommit?.replace) { + this.snapshot = this.pendingCommit.next + return this.retryPending() } + this.snapshot = next + const base = this.pendingCommit?.previous ?? previous + const operation = {previous: base, next} + this.pendingCommit = operation + this.cancelRetry() + return this.persistOperation(operation, 'commit') } - retryPending(): SessionStorageCommitResult { - if (!this.pendingSnapshot) return {status: 'committed'} - const next = this.pendingSnapshot - try { - writeToStorage(next) - this.pendingSnapshot = undefined - this.cancelRetry() - this.broadcast.postMessage({event: UPDATE_EVENT}) - return {status: 'committed'} - } catch (cause) { - const error = storageError('retry', cause) - logStorageError(error) - this.scheduleRetry() - return {status: 'pending', error} - } + async retryPending(): Promise { + if (!this.pendingCommit) return {status: 'committed'} + return this.persistOperation(this.pendingCommit, 'retry') } subscribe(listener: (snapshot: SessionSnapshot) => void): () => void { @@ -98,18 +113,17 @@ export class WebSessionRepository implements SessionRepository { } } - // eslint-disable-next-line @typescript-eslint/require-await async clear(): Promise { - try { - writeToStorage(EMPTY_SNAPSHOT) - this.snapshot = EMPTY_SNAPSHOT - this.pendingSnapshot = undefined - this.cancelRetry() - this.broadcast.postMessage({event: UPDATE_EVENT}) - } catch (cause) { - const error = storageError('clear', cause) - logStorageError(error) - throw cause + this.snapshot = EMPTY_SNAPSHOT + this.pendingCommit = { + previous: EMPTY_SNAPSHOT, + next: EMPTY_SNAPSHOT, + replace: true, + } + this.cancelRetry() + const result = await this.persistOperation(this.pendingCommit, 'clear') + if (result.status === 'pending') { + throw new Error(`session storage clear failed: ${result.error.kind}`) } } @@ -131,11 +145,16 @@ export class WebSessionRepository implements SessionRepository { private receiveExternalUpdate() { try { const next = readFromStorage() + if (this.pendingCommit) { + if (!this.activeOperations.has(this.pendingCommit)) { + void this.retryPending() + } + return + } if (!next || JSON.stringify(next) === JSON.stringify(this.snapshot)) { return } this.snapshot = next - this.pendingSnapshot = undefined this.cancelRetry() this.listeners.forEach(listener => listener(next)) } catch (cause) { @@ -147,7 +166,7 @@ export class WebSessionRepository implements SessionRepository { if (this.retryTimer) return this.retryTimer = setTimeout(() => { this.retryTimer = undefined - this.retryPending() + void this.retryPending() }, RETRY_DELAY) } @@ -155,27 +174,237 @@ export class WebSessionRepository implements SessionRepository { if (this.retryTimer) clearTimeout(this.retryTimer) this.retryTimer = undefined } + + private attachListeners() { + if (this.opened) return + this.opened = true + this.broadcast.onmessage = this.onBroadcastMessage + window.addEventListener('storage', this.onStorage) + } + + private async persistInitialSnapshot( + snapshot: SessionSnapshot, + ): Promise { + const operation = {previous: snapshot, next: snapshot} + this.pendingCommit = operation + const result = await this.persistOperation(operation, 'open') + return result.status === 'committed' + } + + private async persistOperation( + operation: PendingCommit, + errorOperation: SessionStorageError['operation'], + ): Promise { + const active = this.activeOperations.get(operation) + if (active) return active + + const promise = (async (): Promise => { + try { + const committed = operation.replace + ? await this.persistReplacement(operation.next) + : await this.persistCommit( + operation.previous, + operation.next, + errorOperation, + ) + this.broadcast.postMessage({event: UPDATE_EVENT}) + if (this.pendingCommit === operation) { + this.snapshot = committed + this.pendingCommit = undefined + this.cancelRetry() + if (JSON.stringify(committed) !== JSON.stringify(operation.next)) { + this.listeners.forEach(listener => listener(committed)) + } + if (this.legacyMigrationPending) { + this.legacyMigrationPending = false + this.onLegacyMigrationComplete?.() + } + } + return {status: 'committed'} + } catch (cause) { + const error = storageError(errorOperation, cause) + logStorageError(error) + if (this.pendingCommit === operation) this.scheduleRetry() + return {status: 'pending', error} + } finally { + this.activeOperations.delete(operation) + } + })() + this.activeOperations.set(operation, promise) + return promise + } + + private persistCommit( + previous: SessionSnapshot, + next: SessionSnapshot, + errorOperation: SessionStorageError['operation'], + ): Promise { + return withStorageLock(() => { + let latest: SessionSnapshot + try { + latest = readFromStorage() ?? previous + } catch (cause) { + if (!(cause instanceof InvalidWebSessionStorageDataError)) throw cause + logStorageError(storageError(errorOperation, cause)) + latest = previous + } + const committed = rebaseSessionSnapshot(previous, next, latest) + writeToStorage(committed) + return committed + }) + } + + private persistReplacement(next: SessionSnapshot): Promise { + return withStorageLock(() => { + writeToStorage(next) + return next + }) + } +} + +function withStorageLock(callback: () => T): Promise { + const locks = globalThis.navigator?.locks + if (locks) return locks.request(STORAGE_LOCK_NAME, callback) + + let result: T + return update( + STORAGE_LOCK_NAME, + previous => { + result = callback() + return !previous + }, + fallbackLockStore, + ).then(() => result!) } function readFromStorage(): SessionSnapshot | undefined { const raw = localStorage.getItem(STORAGE_KEY) - return raw ? sessionSnapshotSchema.parse(JSON.parse(raw)) : undefined + if (!raw) return undefined + try { + return sessionSnapshotSchema.parse(JSON.parse(raw)) + } catch { + throw new InvalidWebSessionStorageDataError() + } } function writeToStorage(snapshot: SessionSnapshot) { localStorage.setItem(STORAGE_KEY, JSON.stringify(snapshot)) } +class InvalidWebSessionStorageDataError extends Error {} + +const ACCOUNT_KEYS = [ + 'service', + 'did', + 'handle', + 'email', + 'emailConfirmed', + 'emailAuthFactor', + 'refreshJwt', + 'accessJwt', + 'signupQueued', + 'active', + 'status', + 'pdsUrl', + 'isSelfHosted', +] as const satisfies readonly (keyof SessionAccount)[] + +export function rebaseSessionSnapshot( + previous: SessionSnapshot, + next: SessionSnapshot, + latest: SessionSnapshot, +): SessionSnapshot { + const previousByDid = new Map( + previous.accounts.map(account => [account.did, account]), + ) + const nextByDid = new Map( + next.accounts.map(account => [account.did, account]), + ) + const resultByDid = new Map( + latest.accounts.map(account => [account.did, account]), + ) + + for (const did of previousByDid.keys()) { + if (!nextByDid.has(did)) resultByDid.delete(did) + } + for (const account of next.accounts) { + const prior = previousByDid.get(account.did) + if (!prior) { + const latestAccount = resultByDid.get(account.did) + const mergedAccount = latestAccount + ? { + ...latestAccount, + ...account, + refreshJwt: + latestAccount.refreshJwt && account.refreshJwt + ? account.refreshJwt + : undefined, + accessJwt: + latestAccount.accessJwt && account.accessJwt + ? account.accessJwt + : undefined, + } + : account + resultByDid.set(account.did, sessionAccountSchema.parse(mergedAccount)) + continue + } + if (JSON.stringify(prior) === JSON.stringify(account)) continue + const latestAccount = resultByDid.get(account.did) + if (!latestAccount) { + // A concurrent removal wins over an edit based on the removed account. + continue + } + let rebasedAccount = latestAccount + for (const key of ACCOUNT_KEYS) { + if (prior[key] !== account[key]) { + const isCredential = key === 'refreshJwt' || key === 'accessJwt' + const remotelyRevoked = + isCredential && + prior[key] !== latestAccount[key] && + !latestAccount[key] + if (!remotelyRevoked) { + rebasedAccount = {...rebasedAccount, [key]: account[key]} + } + } + } + resultByDid.set(account.did, sessionAccountSchema.parse(rebasedAccount)) + } + + const previousOrder = previous.accounts.map(account => account.did) + const nextOrder = next.accounts.map(account => account.did) + const latestOrder = latest.accounts.map(account => account.did) + const latestDids = new Set(latestOrder) + const orderChanged = + JSON.stringify(previousOrder) !== JSON.stringify(nextOrder) + const order = orderChanged + ? [...nextOrder, ...latestOrder.filter(did => !nextByDid.has(did))] + : [...latestOrder, ...nextOrder.filter(did => !latestDids.has(did))] + const accounts = order.flatMap(did => { + const account = resultByDid.get(did) + return account ? [account] : [] + }) + + let currentDid = + previous.currentDid === next.currentDid + ? latest.currentDid + : next.currentDid + if (currentDid && !resultByDid.has(currentDid)) currentDid = undefined + return sessionSnapshotSchema.parse({accounts, currentDid}) +} + function storageError( operation: SessionStorageError['operation'], cause: unknown, ): SessionStorageError { const message = cause instanceof Error ? cause.message : String(cause) - const kind = /quota|disk.*full|storage.*full|no space/i.test(message) - ? 'storage-full' - : operation === 'open' - ? 'unavailable' - : 'write-failed' + const kind = + cause instanceof InvalidWebSessionStorageDataError + ? 'invalid-data' + : /quota|disk.*full|storage.*full|no space/i.test(message) + ? 'storage-full' + : operation === 'open' + ? 'unavailable' + : 'write-failed' return {kind, operation} } diff --git a/src/state/session/storage/types.ts b/src/state/session/storage/types.ts index 6192a711c5..ff51751661 100644 --- a/src/state/session/storage/types.ts +++ b/src/state/session/storage/types.ts @@ -26,14 +26,19 @@ export type SessionStorageCommitResult = | {status: 'committed'} | {status: 'pending'; error: SessionStorageError} +export type MaybePromise = T | Promise + export interface SessionRepository { - open(legacy?: SessionSnapshot): Promise + open( + legacy?: SessionSnapshot, + onLegacyMigrationComplete?: () => void, + ): Promise getSnapshot(): SessionSnapshot commit( previous: SessionSnapshot, next: SessionSnapshot, - ): SessionStorageCommitResult - retryPending(): SessionStorageCommitResult + ): MaybePromise + retryPending(): MaybePromise subscribe(listener: (snapshot: SessionSnapshot) => void): () => void clear(): Promise }