From 3c974a30f21a60cd155a080d72f35fed1a3b0868 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Fri, 14 Aug 2026 16:27:25 +0900 Subject: [PATCH 1/2] feat(kit): add offline mutation controls and local reset --- projects/kit/README.md | 65 +++++ .../lib/offline-community-sqlite-config.ts | 4 + .../lib/offline-coordinator.service.spec.ts | 32 ++- .../src/lib/offline-coordinator.service.ts | 10 +- .../offline/src/lib/offline-kit-options.ts | 28 ++- .../src/lib/offline-local-reset.spec.ts | 208 ++++++++++++++++ .../offline/src/lib/offline-local-reset.ts | 118 +++++++++ .../lib/offline-mutation-admission.service.ts | 53 ++++ ...fline-mutation-persistence.service.spec.ts | 230 ++++++++++++++++++ .../offline-mutation-persistence.service.ts | 171 +++++++++++++ .../kit/offline/src/lib/offline-provider.ts | 17 ++ .../src/lib/offline-sync.service.spec.ts | 24 ++ .../offline/src/lib/offline-sync.service.ts | 36 +-- .../src/lib/offline.interceptor.spec.ts | 40 ++- .../offline/src/lib/offline.interceptor.ts | 12 +- .../src/lib/sqlite-offline-repository.ts | 14 +- projects/kit/offline/src/public-api.ts | 7 + 17 files changed, 1030 insertions(+), 39 deletions(-) create mode 100644 projects/kit/offline/src/lib/offline-community-sqlite-config.ts create mode 100644 projects/kit/offline/src/lib/offline-local-reset.spec.ts create mode 100644 projects/kit/offline/src/lib/offline-local-reset.ts create mode 100644 projects/kit/offline/src/lib/offline-mutation-admission.service.ts create mode 100644 projects/kit/offline/src/lib/offline-mutation-persistence.service.spec.ts create mode 100644 projects/kit/offline/src/lib/offline-mutation-persistence.service.ts diff --git a/projects/kit/README.md b/projects/kit/README.md index 7926115..b4bd3b0 100644 --- a/projects/kit/README.md +++ b/projects/kit/README.md @@ -495,6 +495,43 @@ provideOffline({ }); ``` +Products may make durable mutation saving device-configurable without disabling replica reads. Supply only the +durable preference adapter; Kit owns the concurrency-sensitive transition. It starts admission closed while loading, +gates every `enqueue` / `enqueuePrepared` / `enqueuePreparedBatch`, and routes matched HTTP writes to normal transport +while disabled. Disabling closes admission synchronously, waits for already accepted commits, flushes pending +commands, verifies an empty Outbox, and then persists `false`. A failed transition restores the last successfully +persisted state. Replacement and discard APIs remain available because they resolve existing commands rather than +create additional pending work. + +```ts +@Injectable({ providedIn: 'root' }) +class ProductMutationPersistenceAdapter implements OfflineMutationPersistenceAdapter { + readonly #settings = inject(ProductSettingsService); + + loadEnabled(): Promise { + return this.#settings.get('offlineMutationPersistence'); + } + + saveEnabled(enabled: boolean): Promise { + return this.#settings.set('offlineMutationPersistence', enabled); + } +} + +provideOffline({ + // repository/schema/executor/puller options omitted + mutationPersistence: { + adapter: ProductMutationPersistenceAdapter, + defaultEnabled: true, + }, +}); + +const offline = inject(OfflineCoordinatorService); +await offline.mutationPersistence.setEnabled(false); +``` + +The product owns settings UI, labels, confirmation copy, and the storage key. Omitting `mutationPersistence` keeps +the historical always-enabled behavior. `readCacheOnly` applications do not expose this setting. + The native offline runtime uses `@capacitor-community/sqlite` on iOS and Android. Install the plugin in the app and sync native projects: @@ -551,6 +588,34 @@ provideOffline({ }); ``` +An explicit user-requested local reset must run before Angular and Kit initialize SQLite. Use the cold-start helpers +instead of reproducing the community plugin's encrypted connection lifecycle in every application. Kit never invokes +reset automatically after a storage error. The product must show destructive confirmation first and owns the marker key. +Only databases that use Kit's same encrypted secret-mode, connection-version-1, read/write lifecycle belong in +`kitCompatibleDatabaseNames`; delete differently configured databases and files in `additionalCleanup`. The marker is +removed only after every database and product cleanup succeeds, so a partial failure is retried on the next cold launch. + +```ts +// Settings action +await requestOfflineLocalReset({ + markerStore: Preferences, + markerKey: 'product:offline:reset', +}); + +// main.ts, before bootstrapApplication(...). Report reset failure but always continue startup. +await recoverOfflineLocalReset({ + markerStore: Preferences, + markerKey: 'product:offline:reset', + sqliteConnection, + kitCompatibleDatabaseNames: ['product-offline', 'product-offline-media'], + additionalCleanup: removeOfflineMediaFiles, +}).catch((error: unknown) => { + console.error('Offline local reset failed; it will retry on the next launch.', error); +}); + +await bootstrapApplication(AppComponent, appConfig); +``` + Replica identity follows the product database: | Identity declaration | SQLite primary key | Row / executor identity | diff --git a/projects/kit/offline/src/lib/offline-community-sqlite-config.ts b/projects/kit/offline/src/lib/offline-community-sqlite-config.ts new file mode 100644 index 0000000..53e1cc1 --- /dev/null +++ b/projects/kit/offline/src/lib/offline-community-sqlite-config.ts @@ -0,0 +1,4 @@ +export const COMMUNITY_SQLITE_ENCRYPTED = true; +export const COMMUNITY_SQLITE_MODE = 'secret'; +export const COMMUNITY_SQLITE_VERSION = 1; +export const COMMUNITY_SQLITE_READONLY = false; diff --git a/projects/kit/offline/src/lib/offline-coordinator.service.spec.ts b/projects/kit/offline/src/lib/offline-coordinator.service.spec.ts index e01f1d0..6063a89 100644 --- a/projects/kit/offline/src/lib/offline-coordinator.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-coordinator.service.spec.ts @@ -1,9 +1,10 @@ -import { signal } from '@angular/core'; +import { ErrorHandler, signal } from '@angular/core'; import { TestBed } from '@angular/core/testing'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { OfflineCoordinatorService } from './offline-coordinator.service'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; import { OfflineNetworkService } from './offline-network.service'; +import { OFFLINE_MUTATION_PERSISTENCE_ADAPTER } from './offline-mutation-persistence.service'; import { OFFLINE_REPOSITORY, OFFLINE_SCHEMA_VERSION, type OfflineScope } from './offline-repository'; import { OfflineSessionService, type OfflineSessionManifest } from './offline-session.service'; import { OfflineStorageUnavailableError } from './offline-storage'; @@ -20,9 +21,11 @@ describe('OfflineCoordinatorService', () => { options: { repositoryInitialize?: () => Promise; onStorageUnavailable?: (error: OfflineStorageUnavailableError) => void | Promise; + preferenceLoad?: () => Promise; } = {}, ) { const order: string[] = []; + const handleError = vi.fn(); const sessionState: { userId: number | null } = { userId: null }; const repository = { initialize: vi.fn(options.repositoryInitialize ?? (async () => undefined)), @@ -64,6 +67,11 @@ describe('OfflineCoordinatorService', () => { discardAllPending: vi.fn(async () => undefined), flush: vi.fn(async () => undefined), }; + class TestMutationPersistenceAdapter { + loadEnabled = options.preferenceLoad ?? (async () => true); + saveEnabled = vi.fn(async () => undefined); + } + const mutationPersistence = options.preferenceLoad ? { adapter: TestMutationPersistenceAdapter } : undefined; TestBed.configureTestingModule({ providers: [ OfflineCoordinatorService, @@ -71,12 +79,17 @@ describe('OfflineCoordinatorService', () => { { provide: OfflineNetworkService, useValue: network }, { provide: OfflineSessionService, useValue: session }, { provide: OfflineSyncService, useValue: sync }, + { provide: ErrorHandler, useValue: { handleError } }, + ...(mutationPersistence + ? [TestMutationPersistenceAdapter, { provide: OFFLINE_MUTATION_PERSISTENCE_ADAPTER, useExisting: TestMutationPersistenceAdapter }] + : []), { provide: OFFLINE_KIT_OPTIONS, useValue: { databaseName: 'test-offline', replicaSchema: emptyReplicaSchema, onStorageUnavailable: options.onStorageUnavailable, + mutationPersistence, }, }, ], @@ -89,9 +102,26 @@ describe('OfflineCoordinatorService', () => { sync, network, repository, + handleError, }; } + it('continues repository, network, session, and sync startup with mutation admission fail-closed after preference read failure', async () => { + const failure = new Error('preference unavailable'); + const { coordinator, repository, network, session, sync, handleError } = setup(null, { + preferenceLoad: async () => Promise.reject(failure), + }); + + await expect(coordinator.initialize()).resolves.toBeUndefined(); + + expect(coordinator.mutationPersistence.enabled()).toBe(false); + expect(handleError).toHaveBeenCalledExactlyOnceWith(failure); + expect(repository.initialize).toHaveBeenCalledOnce(); + expect(network.initialize).toHaveBeenCalledOnce(); + expect(session.initialize).toHaveBeenCalledOnce(); + expect(sync.initialize).toHaveBeenCalledOnce(); + }); + it('restores local visibility without starting remote synchronization', async () => { const manifest = { userId: 1, scopeIds: ['2'], authSubject: 'subject', updatedAt: 1 }; const { coordinator, order, sync } = setup(manifest); diff --git a/projects/kit/offline/src/lib/offline-coordinator.service.ts b/projects/kit/offline/src/lib/offline-coordinator.service.ts index 8dc1fe2..c90f7c8 100644 --- a/projects/kit/offline/src/lib/offline-coordinator.service.ts +++ b/projects/kit/offline/src/lib/offline-coordinator.service.ts @@ -2,6 +2,7 @@ import { computed, inject, Injectable, signal } from '@angular/core'; import type { OfflinePrincipalId } from './offline-identity'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; import { OfflineNetworkService } from './offline-network.service'; +import { OfflineMutationPersistenceService } from './offline-mutation-persistence.service'; import { OFFLINE_REPOSITORY } from './offline-repository'; import { OfflineSessionService } from './offline-session.service'; import type { OfflineSessionManifest, OfflineSessionTransitionLease } from './offline-session.service'; @@ -23,6 +24,7 @@ export interface OfflineResumeRemoteSessionOptions { export class OfflineCoordinatorService { readonly #repository = inject(OFFLINE_REPOSITORY); readonly #network = inject(OfflineNetworkService); + readonly #mutationPersistence = inject(OfflineMutationPersistenceService); readonly #sync = inject(OfflineSyncService); readonly #session = inject(OfflineSessionService); readonly #options = inject(OFFLINE_KIT_OPTIONS); @@ -47,6 +49,8 @@ export class OfflineCoordinatorService { readonly syncState = this.#sync.syncState; readonly pendingCount = this.#sync.pendingCount; readonly conflicts = this.#sync.conflicts; + /** Device-local control for accepting new durable Outbox mutations. */ + readonly mutationPersistence = this.#mutationPersistence; /** * Opens local storage, then session and sync. @@ -56,6 +60,7 @@ export class OfflineCoordinatorService { * resolves with {@link storageState} `unavailable` and skips session/sync initialization. */ async initialize(): Promise { + await this.#mutationPersistence.initialize(); const networkReady = this.#network.initialize(); const initializeRepository = async (): Promise => this.#repository.initialize(); const repositoryReady = await initializeRepository().then( @@ -109,7 +114,10 @@ export class OfflineCoordinatorService { /** * Activates a restored identity for local replica/outbox use without enabling transport sync. */ - async activateOfflineSession(authSubject?: string | null, authLease?: OfflineSessionTransitionLease): Promise { + async activateOfflineSession( + authSubject?: string | null, + authLease?: OfflineSessionTransitionLease, + ): Promise { if (this.#storageUnavailable()) return null; const revision = ++this.#transitionRevision; const lease = this.#lease(revision, authLease); diff --git a/projects/kit/offline/src/lib/offline-kit-options.ts b/projects/kit/offline/src/lib/offline-kit-options.ts index 1d6b309..9018b39 100644 --- a/projects/kit/offline/src/lib/offline-kit-options.ts +++ b/projects/kit/offline/src/lib/offline-kit-options.ts @@ -1,4 +1,4 @@ -import { InjectionToken } from '@angular/core'; +import { InjectionToken, type Type } from '@angular/core'; import type { OfflineReplicaSchemaBundle } from './offline-replica-schema'; import type { OfflineStorageUnavailableError } from './offline-storage'; @@ -10,6 +10,22 @@ export interface OfflineOutboxLimits { maxBytesPerUser?: number; } +/** Product persistence adapter for the device-local offline mutation preference. */ +export interface OfflineMutationPersistenceAdapter { + /** Loads the last durable preference. `null` or `undefined` uses the configured default. */ + loadEnabled(): Promise; + /** Persists a completed enable or disable transition. */ + saveEnabled(enabled: boolean): Promise; +} + +/** Configuration for device-local mutation persistence control. */ +export interface OfflineMutationPersistenceOptions { + /** Injectable product adapter for the durable preference store. */ + adapter: Type; + /** Initial value when no durable preference exists. Defaults to `true`. */ + defaultEnabled?: boolean; +} + /** Product-independent native offline persistence settings. */ export interface OfflineKitOptions { /** Runtime transport mode. Defaults to full synchronized replica/outbox behavior. */ @@ -30,6 +46,13 @@ export interface OfflineKitOptions { wireProtocol?: OfflineWireProtocolFingerprint; /** Optional durable Outbox backpressure policy. */ outboxLimits?: OfflineOutboxLimits; + /** + * Optional device-local control for accepting new durable Outbox mutations. + * + * Replica reads remain enabled while mutation persistence is disabled. When omitted, + * Kit preserves the historical always-enabled mutation behavior. + */ + mutationPersistence?: OfflineMutationPersistenceOptions; /** * Optional product callback invoked when local storage initialization fails. * @@ -39,7 +62,8 @@ export interface OfflineKitOptions { * When omitted, initialization throws {@link OfflineStorageUnavailableError} as before. * If the callback throws or rejects, initialization still fails. * - * Kit never deletes Outbox or replica data in response to storage failure; any reset is product-owned. + * Kit never deletes Outbox or replica data in response to storage failure. Products may invoke + * `recoverOfflineLocalReset` only after an explicit destructive reset request. */ onStorageUnavailable?: (error: OfflineStorageUnavailableError) => void | Promise; } diff --git a/projects/kit/offline/src/lib/offline-local-reset.spec.ts b/projects/kit/offline/src/lib/offline-local-reset.spec.ts new file mode 100644 index 0000000..145e436 --- /dev/null +++ b/projects/kit/offline/src/lib/offline-local-reset.spec.ts @@ -0,0 +1,208 @@ +import { describe, expect, it, vi } from 'vitest'; +import { + recoverOfflineLocalReset, + requestOfflineLocalReset, + type OfflineLocalResetMarkerStore, + type OfflineLocalResetSqliteConnection, +} from './offline-local-reset'; + +describe('offline local reset', () => { + it('persists the marker before reloading', async () => { + const events: string[] = []; + const markerStore = store({ + set: vi.fn(async () => { + events.push('marker'); + }), + }); + + await requestOfflineLocalReset({ + markerStore, + markerKey: 'product:offline-reset', + reloadTarget: { reload: () => events.push('reload') }, + }); + + expect(events).toEqual(['marker', 'reload']); + expect(markerStore.set).toHaveBeenCalledWith({ key: 'product:offline-reset', value: 'requested' }); + }); + + it('deletes the Kit database and product data before removing the marker', async () => { + const events: string[] = []; + const markerStore = store({ + get: vi.fn(async () => ({ value: 'requested' })), + remove: vi.fn(async () => { + events.push('remove-marker'); + }), + }); + const sqlite = connection(events); + + await expect( + recoverOfflineLocalReset({ + markerStore, + markerKey: 'product:offline-reset', + sqliteConnection: sqlite, + kitCompatibleDatabaseNames: ['product-offline', 'product-media'], + nativePlatform: true, + additionalCleanup: async () => { + events.push('product-cleanup'); + }, + }), + ).resolves.toBe(true); + + expect(events).toEqual([ + 'consistency', + 'is-database', + 'create', + 'delete', + 'close', + 'is-database', + 'create', + 'delete', + 'close', + 'product-cleanup', + 'remove-marker', + ]); + expect(sqlite.createConnection).toHaveBeenNthCalledWith(1, 'product-offline', true, 'secret', 1, false); + expect(sqlite.createConnection).toHaveBeenNthCalledWith(2, 'product-media', true, 'secret', 1, false); + }); + + it('does nothing outside native or without a requested marker', async () => { + const markerStore = store(); + const sqlite = connection([]); + + await expect( + recoverOfflineLocalReset({ + markerStore, + markerKey: 'product:offline-reset', + sqliteConnection: sqlite, + kitCompatibleDatabaseNames: ['product-offline'], + nativePlatform: false, + }), + ).resolves.toBe(false); + expect(markerStore.get).not.toHaveBeenCalled(); + + await expect( + recoverOfflineLocalReset({ + markerStore, + markerKey: 'product:offline-reset', + sqliteConnection: sqlite, + kitCompatibleDatabaseNames: ['product-offline'], + nativePlatform: true, + }), + ).resolves.toBe(false); + expect(sqlite.checkConnectionsConsistency).not.toHaveBeenCalled(); + }); + + it('retains the marker when database deletion fails and still closes the connection', async () => { + const markerStore = store({ get: vi.fn(async () => ({ value: 'requested' })) }); + const sqlite = connection([]); + const failure = new Error('delete failed'); + vi.mocked(await sqlite.createConnection('unused', true, 'secret', 1, false)).delete.mockRejectedValueOnce(failure); + vi.mocked(sqlite.createConnection).mockClear(); + + await expect( + recoverOfflineLocalReset({ + markerStore, + markerKey: 'product:offline-reset', + sqliteConnection: sqlite, + kitCompatibleDatabaseNames: ['product-offline'], + nativePlatform: true, + }), + ).rejects.toBe(failure); + + expect(sqlite.closeConnection).toHaveBeenCalledWith('product-offline', false); + expect(markerStore.remove).not.toHaveBeenCalled(); + }); + + it('retains the marker and reports close failure after successful deletion', async () => { + const markerStore = store({ get: vi.fn(async () => ({ value: 'requested' })) }); + const sqlite = connection([]); + const failure = new Error('close failed'); + vi.mocked(sqlite.closeConnection).mockRejectedValueOnce(failure); + + await expect( + recoverOfflineLocalReset({ + markerStore, + markerKey: 'product:offline-reset', + sqliteConnection: sqlite, + kitCompatibleDatabaseNames: ['product-offline'], + nativePlatform: true, + }), + ).rejects.toBe(failure); + + expect(markerStore.remove).not.toHaveBeenCalled(); + }); + + it('preserves both delete and close failures for diagnosis', async () => { + const markerStore = store({ get: vi.fn(async () => ({ value: 'requested' })) }); + const sqlite = connection([]); + const deleteFailure = new Error('delete failed'); + const closeFailure = new Error('close failed'); + vi.mocked(await sqlite.createConnection('unused', true, 'secret', 1, false)).delete.mockRejectedValueOnce(deleteFailure); + vi.mocked(sqlite.createConnection).mockClear(); + vi.mocked(sqlite.closeConnection).mockRejectedValueOnce(closeFailure); + + const reset = recoverOfflineLocalReset({ + markerStore, + markerKey: 'product:offline-reset', + sqliteConnection: sqlite, + kitCompatibleDatabaseNames: ['product-offline'], + nativePlatform: true, + }); + + await expect(reset).rejects.toEqual(expect.objectContaining({ errors: expect.arrayContaining([deleteFailure, closeFailure]) })); + expect(markerStore.remove).not.toHaveBeenCalled(); + }); + + it('retains the marker when product cleanup fails', async () => { + const markerStore = store({ get: vi.fn(async () => ({ value: 'requested' })) }); + const sqlite = connection([]); + const failure = new Error('media cleanup failed'); + + await expect( + recoverOfflineLocalReset({ + markerStore, + markerKey: 'product:offline-reset', + sqliteConnection: sqlite, + kitCompatibleDatabaseNames: ['product-offline'], + nativePlatform: true, + additionalCleanup: async () => Promise.reject(failure), + }), + ).rejects.toBe(failure); + + expect(markerStore.remove).not.toHaveBeenCalled(); + }); + + function store(overrides: Partial = {}): OfflineLocalResetMarkerStore { + return { + get: vi.fn(async () => ({ value: null })), + set: vi.fn(async () => undefined), + remove: vi.fn(async () => undefined), + ...overrides, + }; + } + + function connection(events: string[]): OfflineLocalResetSqliteConnection { + const database = { + delete: vi.fn(async () => { + events.push('delete'); + }), + }; + return { + checkConnectionsConsistency: vi.fn(async () => { + events.push('consistency'); + return { result: true }; + }), + isDatabase: vi.fn(async () => { + events.push('is-database'); + return { result: true }; + }), + createConnection: vi.fn(async () => { + events.push('create'); + return database; + }), + closeConnection: vi.fn(async () => { + events.push('close'); + }), + }; + } +}); diff --git a/projects/kit/offline/src/lib/offline-local-reset.ts b/projects/kit/offline/src/lib/offline-local-reset.ts new file mode 100644 index 0000000..cf166f7 --- /dev/null +++ b/projects/kit/offline/src/lib/offline-local-reset.ts @@ -0,0 +1,118 @@ +import { Capacitor } from '@capacitor/core'; +import { + COMMUNITY_SQLITE_ENCRYPTED, + COMMUNITY_SQLITE_MODE, + COMMUNITY_SQLITE_READONLY, + COMMUNITY_SQLITE_VERSION, +} from './offline-community-sqlite-config'; + +/** Durable marker store used to request a cold-start local reset. */ +export interface OfflineLocalResetMarkerStore { + get(options: { key: string }): Promise<{ value: string | null }>; + set(options: { key: string; value: string }): Promise; + remove(options: { key: string }): Promise; +} + +/** Minimal page reload surface used after a reset request is persisted. */ +export interface OfflineLocalResetReloadTarget { + reload(): void; +} + +/** Native SQLite connection lifecycle required to delete Kit's encrypted database safely. */ +export interface OfflineLocalResetSqliteConnection { + checkConnectionsConsistency(): Promise<{ result?: boolean }>; + isDatabase(database: string): Promise<{ result?: boolean }>; + createConnection( + database: string, + encrypted: boolean, + mode: string, + version: number, + readonly: boolean, + ): Promise<{ delete(): Promise }>; + closeConnection(database: string, readonly: boolean): Promise; +} + +/** Options for {@link requestOfflineLocalReset}. */ +export interface RequestOfflineLocalResetOptions { + markerStore: OfflineLocalResetMarkerStore; + markerKey: string; + reloadTarget?: OfflineLocalResetReloadTarget; +} + +/** Options for {@link recoverOfflineLocalReset}. */ +export interface RecoverOfflineLocalResetOptions { + markerStore: OfflineLocalResetMarkerStore; + markerKey: string; + sqliteConnection: OfflineLocalResetSqliteConnection; + /** + * Kit database followed only by product databases using Kit's same encrypted + * secret-mode, connection-version-1, read/write lifecycle. + * Delete databases with other connection settings in {@link additionalCleanup}. + */ + kitCompatibleDatabaseNames: readonly [string, ...string[]]; + nativePlatform?: boolean; + /** Product-owned cleanup, such as a media database or files, run before the marker is removed. */ + additionalCleanup?: () => Promise; +} + +const OFFLINE_LOCAL_RESET_REQUESTED = 'requested'; + +/** Persists an explicit destructive reset request, then reloads into a cold bootstrap. */ +export async function requestOfflineLocalReset(options: RequestOfflineLocalResetOptions): Promise { + await options.markerStore.set({ key: options.markerKey, value: OFFLINE_LOCAL_RESET_REQUESTED }); + const reloadTarget = options.reloadTarget ?? globalThis.location; + reloadTarget.reload(); +} + +/** + * Recovers an explicit reset request before Angular and Kit open native SQLite. + * + * The marker is removed only after the Kit database and every product cleanup succeed. + * Kit never invokes this helper automatically in response to a storage failure. + */ +export async function recoverOfflineLocalReset(options: RecoverOfflineLocalResetOptions): Promise { + if (!(options.nativePlatform ?? Capacitor.isNativePlatform())) return false; + const marker = await options.markerStore.get({ key: options.markerKey }); + if (marker.value !== OFFLINE_LOCAL_RESET_REQUESTED) return false; + + await options.sqliteConnection.checkConnectionsConsistency(); + for (const databaseName of new Set(options.kitCompatibleDatabaseNames)) { + const exists = await options.sqliteConnection.isDatabase(databaseName); + if (exists.result) { + const connection = await options.sqliteConnection.createConnection( + databaseName, + COMMUNITY_SQLITE_ENCRYPTED, + COMMUNITY_SQLITE_MODE, + COMMUNITY_SQLITE_VERSION, + COMMUNITY_SQLITE_READONLY, + ); + await deleteAndCloseOfflineDatabase(options.sqliteConnection, databaseName, connection); + } + } + await options.additionalCleanup?.(); + await options.markerStore.remove({ key: options.markerKey }); + return true; +} + +type OfflineResetOperationResult = { ok: true } | { ok: false; error: unknown }; + +async function settleOfflineResetOperation(operation: () => Promise): Promise { + return new Promise((resolve) => resolve(operation())).then( + () => ({ ok: true }), + (error: unknown) => ({ ok: false, error }), + ); +} + +async function deleteAndCloseOfflineDatabase( + sqliteConnection: OfflineLocalResetSqliteConnection, + databaseName: string, + connection: { delete(): Promise }, +): Promise { + const deletion = await settleOfflineResetOperation(() => connection.delete()); + const closing = await settleOfflineResetOperation(() => sqliteConnection.closeConnection(databaseName, COMMUNITY_SQLITE_READONLY)); + if (!deletion.ok && !closing.ok) { + throw new AggregateError([deletion.error, closing.error], `Offline database ${databaseName} delete and close both failed.`); + } + if (!deletion.ok) throw deletion.error; + if (!closing.ok) throw closing.error; +} diff --git a/projects/kit/offline/src/lib/offline-mutation-admission.service.ts b/projects/kit/offline/src/lib/offline-mutation-admission.service.ts new file mode 100644 index 0000000..ce39bce --- /dev/null +++ b/projects/kit/offline/src/lib/offline-mutation-admission.service.ts @@ -0,0 +1,53 @@ +import { inject, Injectable } from '@angular/core'; +import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; + +/** Raised when a new durable mutation reaches Kit after mutation persistence was closed. */ +export class OfflineMutationPersistenceDisabledError extends Error { + constructor() { + super('Offline mutation persistence was disabled before the mutation could be accepted.'); + this.name = 'OfflineMutationPersistenceDisabledError'; + } +} + +/** Product-independent lease gate around every new durable Outbox command. */ +@Injectable({ providedIn: 'root' }) +export class OfflineMutationAdmissionService { + readonly #options = inject(OFFLINE_KIT_OPTIONS); + #accepting = this.#options.mutationPersistence === undefined; + #active = 0; + #idle: Promise | null = null; + #resolveIdle: (() => void) | null = null; + + get accepting(): boolean { + return this.#accepting; + } + + async run(operation: () => Promise): Promise { + if (!this.#accepting) throw new OfflineMutationPersistenceDisabledError(); + this.#active += 1; + return new Promise((resolve) => resolve(operation())).finally(() => this.#release()); + } + + open(): void { + this.#accepting = true; + } + + async close(): Promise { + this.#accepting = false; + if (this.#active === 0) return; + if (!this.#idle) { + this.#idle = new Promise((resolve) => { + this.#resolveIdle = resolve; + }); + } + return this.#idle; + } + + #release(): void { + this.#active -= 1; + if (this.#active !== 0) return; + this.#resolveIdle?.(); + this.#idle = null; + this.#resolveIdle = null; + } +} diff --git a/projects/kit/offline/src/lib/offline-mutation-persistence.service.spec.ts b/projects/kit/offline/src/lib/offline-mutation-persistence.service.spec.ts new file mode 100644 index 0000000..03c5261 --- /dev/null +++ b/projects/kit/offline/src/lib/offline-mutation-persistence.service.spec.ts @@ -0,0 +1,230 @@ +import { ErrorHandler, signal } from '@angular/core'; +import { TestBed } from '@angular/core/testing'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { OFFLINE_KIT_OPTIONS, type OfflineKitOptions } from './offline-kit-options'; +import { OfflineMutationAdmissionService, OfflineMutationPersistenceDisabledError } from './offline-mutation-admission.service'; +import { + OFFLINE_MUTATION_PERSISTENCE_ADAPTER, + OfflineMutationPersistencePendingError, + OfflineMutationPersistenceRequiresOnlineError, + OfflineMutationPersistenceService, +} from './offline-mutation-persistence.service'; +import { OfflineNetworkService } from './offline-network.service'; +import { OfflineSyncService } from './offline-sync.service'; + +describe('OfflineMutationPersistenceService', () => { + const networkState = signal<'online' | 'offline'>('online'); + const pendingCount = signal(0); + const flush = vi.fn(async () => undefined); + const handleError = vi.fn(); + + beforeEach(() => { + TestBed.resetTestingModule(); + networkState.set('online'); + pendingCount.set(0); + flush.mockReset(); + handleError.mockReset(); + }); + + it('keeps historical always-enabled admission when no preference adapter is configured', async () => { + const { service, admission } = setup({ databaseName: 'test', replicaSchema: {} as never }); + + await service.initialize(); + + expect(service.available).toBe(false); + expect(service.enabled()).toBe(true); + await expect(admission.run(async () => 'accepted')).resolves.toBe('accepted'); + }); + + it('loads a durable OFF preference before opening admission', async () => { + const loadEnabled = vi.fn(async () => false); + const { service, admission } = setup(options(loadEnabled, vi.fn())); + + await service.initialize(); + + expect(service.enabled()).toBe(false); + await expect(admission.run(async () => undefined)).rejects.toBeInstanceOf(OfflineMutationPersistenceDisabledError); + }); + + it('uses the configured default when no durable preference exists', async () => { + const { service } = setup(options(async () => null, vi.fn(), false)); + + await service.initialize(); + + expect(service.enabled()).toBe(false); + }); + + it('reports preference loading failure and continues with admission fail-closed', async () => { + const failure = new Error('settings unavailable'); + const { service, admission } = setup(options(async () => Promise.reject(failure), vi.fn())); + + await expect(service.initialize()).resolves.toBeUndefined(); + expect(service.enabled()).toBe(false); + expect(handleError).toHaveBeenCalledExactlyOnceWith(failure); + await expect(admission.run(async () => undefined)).rejects.toBeInstanceOf(OfflineMutationPersistenceDisabledError); + }); + + it('applies a latest enable after a deferred initial OFF load without state/admission divergence', async () => { + let releaseLoad!: (enabled: boolean) => void; + const loadEnabled = vi.fn(() => new Promise((resolve) => (releaseLoad = resolve))); + const saveEnabled = vi.fn(async () => undefined); + const { service, admission } = setup(options(loadEnabled, saveEnabled)); + + const initializing = service.initialize(); + const enabling = service.setEnabled(true); + releaseLoad(false); + await Promise.all([initializing, enabling]); + + expect(saveEnabled).toHaveBeenCalledExactlyOnceWith(true); + expect(service.enabled()).toBe(true); + await expect(admission.run(async () => 'accepted')).resolves.toBe('accepted'); + }); + + it('keeps admission closed when latest disable crosses a deferred initial ON load', async () => { + let releaseLoad!: (enabled: boolean) => void; + const loadEnabled = vi.fn(() => new Promise((resolve) => (releaseLoad = resolve))); + const saveEnabled = vi.fn(async () => undefined); + const { service, admission } = setup(options(loadEnabled, saveEnabled)); + + const initializing = service.initialize(); + const disabling = service.setEnabled(false); + releaseLoad(true); + await Promise.all([initializing, disabling]); + + expect(saveEnabled).toHaveBeenCalledExactlyOnceWith(false); + expect(service.enabled()).toBe(false); + await expect(admission.run(async () => undefined)).rejects.toBeInstanceOf(OfflineMutationPersistenceDisabledError); + }); + + it('applies a latest enable after initial load failure is reported', async () => { + let rejectLoad!: (error: unknown) => void; + const failure = new Error('settings unavailable'); + const loadEnabled = vi.fn(() => new Promise((_resolve, reject) => (rejectLoad = reject))); + const saveEnabled = vi.fn(async () => undefined); + const { service, admission } = setup(options(loadEnabled, saveEnabled)); + + const initializing = service.initialize(); + const enabling = service.setEnabled(true); + rejectLoad(failure); + await Promise.all([initializing, enabling]); + + expect(handleError).toHaveBeenCalledExactlyOnceWith(failure); + expect(saveEnabled).toHaveBeenCalledExactlyOnceWith(true); + expect(service.enabled()).toBe(true); + await expect(admission.run(async () => 'accepted')).resolves.toBe('accepted'); + }); + + it('closes admission, waits for accepted work, flushes, then persists OFF', async () => { + const saveEnabled = vi.fn(async () => undefined); + const { service, admission } = setup(options(async () => true, saveEnabled)); + await service.initialize(); + let release!: () => void; + const accepted = admission.run( + () => + new Promise((resolve) => { + release = resolve; + }), + ); + pendingCount.set(1); + flush.mockImplementationOnce(async () => { + pendingCount.set(0); + }); + + const disabling = service.setEnabled(false); + await expect(admission.run(async () => undefined)).rejects.toBeInstanceOf(OfflineMutationPersistenceDisabledError); + expect(flush).not.toHaveBeenCalled(); + release(); + await accepted; + await disabling; + + expect(flush).toHaveBeenCalledOnce(); + expect(saveEnabled).toHaveBeenCalledExactlyOnceWith(false); + expect(service.enabled()).toBe(false); + }); + + it('reopens admission when an offline disable with pending commands fails', async () => { + const { service, admission } = setup(options(async () => true, vi.fn())); + await service.initialize(); + networkState.set('offline'); + pendingCount.set(1); + + await expect(service.setEnabled(false)).rejects.toBeInstanceOf(OfflineMutationPersistenceRequiresOnlineError); + + expect(service.enabled()).toBe(true); + await expect(admission.run(async () => 'accepted')).resolves.toBe('accepted'); + }); + + it('does not persist OFF when commands remain after flush', async () => { + const saveEnabled = vi.fn(async () => undefined); + const { service } = setup(options(async () => true, saveEnabled)); + await service.initialize(); + pendingCount.set(2); + + await expect(service.setEnabled(false)).rejects.toEqual(new OfflineMutationPersistencePendingError(2)); + + expect(saveEnabled).not.toHaveBeenCalled(); + expect(service.enabled()).toBe(true); + }); + + it('honors a latest disable requested while the preceding enable write is in flight', async () => { + let releaseEnable!: () => void; + const writes: boolean[] = []; + const saveEnabled = vi.fn( + (enabled: boolean) => + new Promise((resolve) => { + writes.push(enabled); + if (enabled) releaseEnable = resolve; + else resolve(); + }), + ); + const { service } = setup(options(async () => false, saveEnabled)); + await service.initialize(); + + const enabling = service.setEnabled(true); + await vi.waitFor(() => expect(writes).toEqual([true])); + const disabling = service.setEnabled(false); + releaseEnable(); + await Promise.all([enabling, disabling]); + + expect(writes).toEqual([true, false]); + expect(service.enabled()).toBe(false); + }); + + function setup(kitOptions: OfflineKitOptions): { + service: OfflineMutationPersistenceService; + admission: OfflineMutationAdmissionService; + } { + const adapter = kitOptions.mutationPersistence?.adapter; + TestBed.configureTestingModule({ + providers: [ + OfflineMutationPersistenceService, + OfflineMutationAdmissionService, + { provide: OFFLINE_KIT_OPTIONS, useValue: kitOptions }, + { provide: OfflineNetworkService, useValue: { state: networkState } }, + { provide: OfflineSyncService, useValue: { pendingCount, flush } }, + { provide: ErrorHandler, useValue: { handleError } }, + ...(adapter ? [adapter, { provide: OFFLINE_MUTATION_PERSISTENCE_ADAPTER, useExisting: adapter }] : []), + ], + }); + return { + service: TestBed.inject(OfflineMutationPersistenceService), + admission: TestBed.inject(OfflineMutationAdmissionService), + }; + } + + function options( + loadEnabled: () => Promise, + saveEnabled: (enabled: boolean) => Promise, + defaultEnabled = true, + ): OfflineKitOptions { + class TestMutationPersistenceAdapter { + loadEnabled = loadEnabled; + saveEnabled = saveEnabled; + } + return { + databaseName: 'test', + replicaSchema: {} as never, + mutationPersistence: { adapter: TestMutationPersistenceAdapter, defaultEnabled }, + }; + } +}); diff --git a/projects/kit/offline/src/lib/offline-mutation-persistence.service.ts b/projects/kit/offline/src/lib/offline-mutation-persistence.service.ts new file mode 100644 index 0000000..422f30c --- /dev/null +++ b/projects/kit/offline/src/lib/offline-mutation-persistence.service.ts @@ -0,0 +1,171 @@ +import { computed, ErrorHandler, inject, Injectable, InjectionToken, signal } from '@angular/core'; +import { OFFLINE_KIT_OPTIONS, type OfflineMutationPersistenceAdapter } from './offline-kit-options'; +import { OfflineMutationAdmissionService } from './offline-mutation-admission.service'; +import { OfflineNetworkService } from './offline-network.service'; +import { OfflineSyncService } from './offline-sync.service'; + +/** Raised when disabling requires pending commands to be synchronized while transport is offline. */ +export class OfflineMutationPersistenceRequiresOnlineError extends Error { + constructor() { + super('Pending offline mutations must be synchronized before persistence can be disabled.'); + this.name = 'OfflineMutationPersistenceRequiresOnlineError'; + } +} + +/** Raised when pending commands remain after the disable transition attempted a flush. */ +export class OfflineMutationPersistencePendingError extends Error { + constructor(readonly pendingCount: number) { + super('Pending offline mutations remain after synchronization.'); + this.name = 'OfflineMutationPersistencePendingError'; + } +} + +type OfflineMutationPersistenceState = 'initializing' | 'enabled' | 'disabling' | 'disabled'; + +const settledMutationPersistence = async (): Promise => undefined; + +function invokeMutationPersistence(operation: () => Promise): Promise { + return new Promise((resolve) => resolve(operation())); +} + +function reportMutationPersistenceError(errorHandler: ErrorHandler, error: unknown): void { + try { + errorHandler.handleError(error); + } catch { + // Preference failure already degraded safely to disabled. Telemetry must not stop bootstrap. + } +} + +/** Read-only mutation admission signal consumed by the HTTP interceptor. */ +export const OFFLINE_MUTATION_PERSISTENCE_ENABLED = new InjectionToken<() => boolean>('OFFLINE_MUTATION_PERSISTENCE_ENABLED', { + factory: () => () => true, +}); + +/** Product adapter used by Kit to persist the device-local mutation preference. */ +export const OFFLINE_MUTATION_PERSISTENCE_ADAPTER = new InjectionToken( + 'OFFLINE_MUTATION_PERSISTENCE_ADAPTER', + { factory: () => null }, +); + +/** Controls whether Kit accepts new durable mutations while leaving replica reads enabled. */ +@Injectable({ providedIn: 'root' }) +export class OfflineMutationPersistenceService { + readonly #options = inject(OFFLINE_KIT_OPTIONS); + readonly #errorHandler = inject(ErrorHandler); + readonly #persistence = inject(OFFLINE_MUTATION_PERSISTENCE_ADAPTER); + readonly #network = inject(OfflineNetworkService); + readonly #sync = inject(OfflineSyncService); + readonly #admission = inject(OfflineMutationAdmissionService); + readonly #configured = this.#options.mode !== 'readCacheOnly' && this.#options.mutationPersistence !== undefined; + readonly #state = signal(this.#configured ? 'initializing' : 'enabled'); + #initializePromise: Promise | null = null; + #transitionTail: Promise = settledMutationPersistence(); + #transitionRevision = 0; + #persistedEnabled = !this.#configured; + #latestTransition: { enabled: boolean; promise: Promise; revision: number } | null = null; + + /** Whether the product configured a device-local mutation persistence preference. */ + readonly available = this.#configured; + /** Whether new durable Outbox mutations are currently accepted. */ + readonly enabled = computed(() => this.#state() === 'enabled'); + /** Whether Kit is draining accepted and pending mutations before disabling. */ + readonly changing = computed(() => this.#state() === 'disabling'); + + /** Loads the durable preference before Kit initializes repository-backed services. */ + async initialize(): Promise { + if (!this.#configured) { + this.#admission.open(); + return; + } + await (this.#initializePromise ??= this.#loadInitialPreference()); + } + + /** + * Enables or disables new durable mutations. + * + * Disabling closes admission synchronously, waits for already admitted commits, + * flushes existing commands, verifies an empty Outbox, and only then persists OFF. + */ + setEnabled(enabled: boolean): Promise { + if (!this.#configured) return settledMutationPersistence(); + const initialization = this.initialize(); + const latest = this.#latestTransition; + if (latest?.enabled === enabled) return latest.promise; + if (!latest && ((enabled && this.#state() === 'enabled') || (!enabled && this.#state() === 'disabled'))) { + return settledMutationPersistence(); + } + + const revision = ++this.#transitionRevision; + if (!enabled) { + this.#state.set('disabling'); + void this.#admission.close(); + } + const transition = this.#transitionTail.then(() => initialization).then(() => this.#applyTransition(enabled, revision)); + this.#transitionTail = transition.catch(() => undefined); + const promise = transition.finally(() => { + if (this.#latestTransition?.revision === revision) this.#latestTransition = null; + }); + this.#latestTransition = { enabled, promise, revision }; + return promise; + } + + async #loadInitialPreference(): Promise { + const persistence = this.#requiredPersistence(); + return invokeMutationPersistence(() => persistence.loadEnabled()).then( + (stored) => { + this.#persistedEnabled = stored ?? this.#options.mutationPersistence?.defaultEnabled ?? true; + if (this.#transitionRevision === 0) { + this.#state.set(this.#persistedEnabled ? 'enabled' : 'disabled'); + if (this.#persistedEnabled) this.#admission.open(); + } + }, + async (error: unknown) => { + this.#persistedEnabled = false; + if (this.#transitionRevision === 0) this.#state.set('disabled'); + await this.#admission.close(); + reportMutationPersistenceError(this.#errorHandler, error); + }, + ); + } + + #applyTransition(enabled: boolean, revision: number): Promise { + if (revision !== this.#transitionRevision) return settledMutationPersistence(); + const transition = enabled ? this.#enable(revision) : this.#disable(revision); + return transition.catch((error: unknown) => { + if (revision === this.#transitionRevision) { + this.#state.set(this.#persistedEnabled ? 'enabled' : 'disabled'); + if (this.#persistedEnabled) this.#admission.open(); + } + return Promise.reject(error); + }); + } + + async #enable(revision: number): Promise { + await this.#requiredPersistence().saveEnabled(true); + this.#persistedEnabled = true; + if (revision !== this.#transitionRevision) return; + this.#state.set('enabled'); + this.#admission.open(); + } + + async #disable(revision: number): Promise { + await this.#admission.close(); + if (revision !== this.#transitionRevision) return; + const pendingBeforeFlush = this.#sync.pendingCount(); + if (pendingBeforeFlush > 0) { + if (this.#network.state() !== 'online') throw new OfflineMutationPersistenceRequiresOnlineError(); + await this.#sync.flush(); + } + if (revision !== this.#transitionRevision) return; + const pendingCount = this.#sync.pendingCount(); + if (pendingCount > 0) throw new OfflineMutationPersistencePendingError(pendingCount); + await this.#requiredPersistence().saveEnabled(false); + this.#persistedEnabled = false; + if (revision === this.#transitionRevision) this.#state.set('disabled'); + } + + #requiredPersistence(): OfflineMutationPersistenceAdapter { + if (!this.#persistence) throw new Error('Offline mutation persistence adapter is not provided.'); + return this.#persistence; + } +} diff --git a/projects/kit/offline/src/lib/offline-provider.ts b/projects/kit/offline/src/lib/offline-provider.ts index 4320333..8ba106d 100644 --- a/projects/kit/offline/src/lib/offline-provider.ts +++ b/projects/kit/offline/src/lib/offline-provider.ts @@ -8,6 +8,11 @@ import { OFFLINE_COMMAND_HOOKS } from './offline-command-hooks'; import type { OfflineKitOptions } from './offline-kit-options'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; import { OfflineCoordinatorService } from './offline-coordinator.service'; +import { + OFFLINE_MUTATION_PERSISTENCE_ADAPTER, + OFFLINE_MUTATION_PERSISTENCE_ENABLED, + OfflineMutationPersistenceService, +} from './offline-mutation-persistence.service'; import { IonicOfflineRepository, OFFLINE_REPOSITORY, @@ -70,6 +75,7 @@ export interface ProvideReadCacheOfflineOptions extends ProvideOfflineOptionsBas /** Creates the encrypted native cache database key on first install. Unused by the web repository. */ createEncryptionKey: () => Promise; mutationPolicies?: never; + mutationPersistence?: never; commandExecutor?: never; replicaPuller?: never; } @@ -116,6 +122,7 @@ export function provideOffline(options: ProvideOfflineOptions): EnvironmentProvi replicaSchema: options.replicaSchema, wireProtocol: options.wireProtocol, outboxLimits: options.outboxLimits, + mutationPersistence: options.mutationPersistence, onStorageUnavailable: options.onStorageUnavailable, }, }, @@ -127,6 +134,16 @@ export function provideOffline(options: ProvideOfflineOptions): EnvironmentProvi provide: OFFLINE_REPOSITORY, useFactory: () => selectOfflineRepository(Capacitor.getPlatform(), inject(IonicOfflineRepository), inject(SqliteOfflineRepository)), }, + { + provide: OFFLINE_MUTATION_PERSISTENCE_ENABLED, + useFactory: () => inject(OfflineMutationPersistenceService).enabled, + }, + ...(options.mutationPersistence + ? [ + options.mutationPersistence.adapter, + { provide: OFFLINE_MUTATION_PERSISTENCE_ADAPTER, useExisting: options.mutationPersistence.adapter }, + ] + : []), { provide: OFFLINE_SYNC_CONTEXT, useExisting: OfflineSessionService }, synchronized ? { provide: OFFLINE_COMMAND_EXECUTOR, useExisting: options.commandExecutor } diff --git a/projects/kit/offline/src/lib/offline-sync.service.spec.ts b/projects/kit/offline/src/lib/offline-sync.service.spec.ts index 1979027..6497c73 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.spec.ts @@ -11,6 +11,7 @@ import { import { OFFLINE_COMMAND_HOOKS } from './offline-command-hooks'; import { OFFLINE_KIT_OPTIONS, type OfflineKitOptions } from './offline-kit-options'; import { OfflineNetworkService } from './offline-network.service'; +import { OfflineMutationAdmissionService, OfflineMutationPersistenceDisabledError } from './offline-mutation-admission.service'; import { OfflineReplicaPullService, OfflineReplicaSchemaMismatchError } from './offline-replica-pull.service'; import { OfflineReplicaMutationCoordinator } from './offline-replica-mutation-coordinator'; import { @@ -376,6 +377,29 @@ describe('OfflineSyncService', () => { expect(rows).toEqual([]); }); + it('rejects every new command entry point after mutation admission closes', async () => { + await TestBed.inject(OfflineMutationAdmissionService).close(); + const request = { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated' as const, localId: 'closed-write' }, + operation: 'documents.create', + payload: { title: 'write' }, + }; + const prepare = vi.fn(async () => ({ request })); + const prepareBatch = vi.fn(async () => [{ request }]); + + await expect(service.enqueue(request, { flush: false })).rejects.toBeInstanceOf(OfflineMutationPersistenceDisabledError); + await expect(service.enqueuePrepared(prepare, { flush: false })).rejects.toBeInstanceOf(OfflineMutationPersistenceDisabledError); + await expect(service.enqueuePreparedBatch(prepareBatch, { flush: false })).rejects.toBeInstanceOf( + OfflineMutationPersistenceDisabledError, + ); + + expect(prepare).not.toHaveBeenCalled(); + expect(prepareBatch).not.toHaveBeenCalled(); + expect(commands).toEqual([]); + }); + it('prepared enqueue rematerializes the base row and declared localOnly footprint', async () => { const companion: OfflineReplicaRow = { userId: 1, diff --git a/projects/kit/offline/src/lib/offline-sync.service.ts b/projects/kit/offline/src/lib/offline-sync.service.ts index 1c403a0..5dc6ef7 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -12,6 +12,7 @@ import { import { OFFLINE_COMMAND_HOOKS } from './offline-command-hooks'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; import { OfflineNetworkService } from './offline-network.service'; +import { OfflineMutationAdmissionService } from './offline-mutation-admission.service'; import { OfflineReplicaPullService, OfflineReplicaSchemaMismatchError } from './offline-replica-pull.service'; import { isOfflineAggregateIntentConflict, offlineAggregateIntentMutations } from './offline-aggregate-intent-projector'; import { commandFootprintKeys, OfflineReplicaMutationCoordinator } from './offline-replica-mutation-coordinator'; @@ -150,6 +151,7 @@ export function offlineRetryDelayMs(attempts: number, random: () => number = Mat @Injectable({ providedIn: 'root' }) export class OfflineSyncService { readonly #network = inject(OfflineNetworkService); + readonly #mutationAdmission = inject(OfflineMutationAdmissionService); readonly #repository = inject(OFFLINE_REPOSITORY); readonly #executor = inject(OFFLINE_COMMAND_EXECUTOR); readonly #context = inject(OFFLINE_SYNC_CONTEXT); @@ -270,8 +272,10 @@ export class OfflineSyncService { } enqueue(request: EnqueueOfflineCommand, options: { flush?: boolean } = {}): Promise { - const generation = this.#generation; - return this.#serializeReplicaMutation((repository) => this.#enqueue(request, options, generation, undefined, repository)); + return this.#mutationAdmission.run(() => { + const generation = this.#generation; + return this.#serializeReplicaMutation((repository) => this.#enqueue(request, options, generation, undefined, repository)); + }); } /** @@ -282,12 +286,14 @@ export class OfflineSyncService { prepare: (repository: OfflineRepository) => Promise>, options: { flush?: boolean } = {}, ): Promise { - const generation = this.#generation; - return this.#serializeReplicaMutation(async (repository) => { - await this.initialize(); - if (!this.#isCurrent(generation)) throw new Error('Offline session changed before prepared enqueue.'); - const prepared = await prepare(repository); - return this.#enqueue(prepared.request, options, generation, undefined, repository); + return this.#mutationAdmission.run(() => { + const generation = this.#generation; + return this.#serializeReplicaMutation(async (repository) => { + await this.initialize(); + if (!this.#isCurrent(generation)) throw new Error('Offline session changed before prepared enqueue.'); + const prepared = await prepare(repository); + return this.#enqueue(prepared.request, options, generation, undefined, repository); + }); }); } @@ -303,12 +309,14 @@ export class OfflineSyncService { prepare: (repository: OfflineRepository) => Promise[]>, options: PreparedOfflineBatchOptions = {}, ): Promise { - const generation = this.#generation; - return this.#serializeReplicaMutation(async (repository) => { - await this.initialize(); - if (!this.#isCurrent(generation)) throw new Error('Offline session changed before prepared batch enqueue.'); - const prepared = await prepare(repository); - return this.#enqueuePreparedBatch(prepared, options, generation, repository); + return this.#mutationAdmission.run(() => { + const generation = this.#generation; + return this.#serializeReplicaMutation(async (repository) => { + await this.initialize(); + if (!this.#isCurrent(generation)) throw new Error('Offline session changed before prepared batch enqueue.'); + const prepared = await prepare(repository); + return this.#enqueuePreparedBatch(prepared, options, generation, repository); + }); }); } diff --git a/projects/kit/offline/src/lib/offline.interceptor.spec.ts b/projects/kit/offline/src/lib/offline.interceptor.spec.ts index 556efe6..beae690 100644 --- a/projects/kit/offline/src/lib/offline.interceptor.spec.ts +++ b/projects/kit/offline/src/lib/offline.interceptor.spec.ts @@ -4,6 +4,7 @@ import { TestBed } from '@angular/core/testing'; import { beforeEach, describe, expect, it, vi } from 'vitest'; import { finalize, firstValueFrom, of, Subject, throwError, type Observable } from 'rxjs'; import { OfflineNetworkService } from './offline-network.service'; +import { OFFLINE_MUTATION_PERSISTENCE_ENABLED } from './offline-mutation-persistence.service'; import { offlineInterceptor } from './offline.interceptor'; import { OFFLINE_BYPASS, @@ -21,6 +22,7 @@ describe('offlineInterceptor', () => { let markApiSuccess: ReturnType; let markApiFailure: ReturnType; let handleError: ReturnType; + let mutationPersistenceEnabled: boolean; beforeEach(() => { resolve = vi.fn(() => null); @@ -28,11 +30,13 @@ describe('offlineInterceptor', () => { markApiSuccess = vi.fn(); markApiFailure = vi.fn(); handleError = vi.fn(); + mutationPersistenceEnabled = true; TestBed.configureTestingModule({ providers: [ { provide: OfflineRequestPolicyRegistry, useValue: { resolve } }, { provide: OfflineMutationRequestPolicyRegistry, useValue: { resolve: resolveMutation } }, { provide: OfflineNetworkService, useValue: { markApiSuccess, markApiFailure } }, + { provide: OFFLINE_MUTATION_PERSISTENCE_ENABLED, useValue: () => mutationPersistenceEnabled }, { provide: ErrorHandler, useValue: { handleError } }, ], }); @@ -173,6 +177,18 @@ describe('offlineInterceptor', () => { expect(markApiFailure).not.toHaveBeenCalled(); }); + it('mutation persistence OFFではmatched policyを解決せずtransportへ渡す', async () => { + mutationPersistenceEnabled = false; + const request = new HttpRequest('POST', '/groups/1/documents', {}); + const response = new HttpResponse({ status: 201 }); + const next = vi.fn(() => of(response)); + + await expect(firstValueFrom(run(request, next))).resolves.toBe(response); + + expect(resolveMutation).not.toHaveBeenCalled(); + expect(next).toHaveBeenCalledOnce(); + }); + it('matched mutationのprepare失敗時はtransportへfall throughしない', async () => { const error = new Error('outbox full'); resolveMutation.mockReturnValue({ @@ -294,9 +310,7 @@ describe('offlineInterceptor', () => { for (const status of [401, 403, 500]) { const error = new HttpErrorResponse({ status }); - await expect( - collect(run(new HttpRequest('GET', '/bootstrap'), () => throwError(() => error))), - ).rejects.toBe(error); + await expect(collect(run(new HttpRequest('GET', '/bootstrap'), () => throwError(() => error)))).rejects.toBe(error); } }); @@ -342,9 +356,9 @@ describe('offlineInterceptor', () => { }), ); - await expect( - collect(run(new HttpRequest('GET', '/bootstrap'), () => of(new HttpResponse({ status: 200 })))), - ).rejects.toBe(projectionError); + await expect(collect(run(new HttpRequest('GET', '/bootstrap'), () => of(new HttpResponse({ status: 200 }))))).rejects.toBe( + projectionError, + ); }); it('remote projectResponseのstatus=0はtransport fallbackとして握りつぶさない', async () => { @@ -360,9 +374,9 @@ describe('offlineInterceptor', () => { }), ); - await expect( - collect(run(new HttpRequest('GET', '/bootstrap'), () => of(new HttpResponse({ status: 200 })))), - ).rejects.toBe(projectionError); + await expect(collect(run(new HttpRequest('GET', '/bootstrap'), () => of(new HttpResponse({ status: 200 }))))).rejects.toBe( + projectionError, + ); }); it('local projectResponse失敗はErrorHandlerへ報告しnetwork-firstへ継続する', async () => { @@ -418,9 +432,11 @@ describe('offlineInterceptor', () => { resolve.mockReturnValue(localFirstPlan({ readLocal: vi.fn(async () => local) })); let transportUnsubscribed = false; const transportSubject = new Subject>(); - const transport$ = transportSubject.asObservable().pipe(finalize(() => { - transportUnsubscribed = true; - })); + const transport$ = transportSubject.asObservable().pipe( + finalize(() => { + transportUnsubscribed = true; + }), + ); const emissions: HttpResponse[] = []; const subscription = run(new HttpRequest('GET', '/bootstrap'), () => transport$).subscribe({ diff --git a/projects/kit/offline/src/lib/offline.interceptor.ts b/projects/kit/offline/src/lib/offline.interceptor.ts index 39bb09d..bee0e75 100644 --- a/projects/kit/offline/src/lib/offline.interceptor.ts +++ b/projects/kit/offline/src/lib/offline.interceptor.ts @@ -20,6 +20,7 @@ import { throwError, } from 'rxjs'; import { isOfflineFallbackError, OfflineNetworkService } from './offline-network.service'; +import { OFFLINE_MUTATION_PERSISTENCE_ENABLED } from './offline-mutation-persistence.service'; import { OFFLINE_BYPASS, OFFLINE_RESPONSE_HEADER, @@ -48,6 +49,7 @@ export const offlineInterceptor: HttpInterceptorFn = (request, next) => { return readNetworkFirst(request, plan, transport, fallback); } if (LOCAL_FIRST_MUTATION_METHODS.has(request.method)) { + if (!inject(OFFLINE_MUTATION_PERSISTENCE_ENABLED)()) return transport(); const plan = inject(OfflineMutationRequestPolicyRegistry).resolve(request); if (plan) { return defer(() => from(plan.prepare())).pipe( @@ -106,10 +108,7 @@ function readLocalFirst( ); } -function resolveLocalAttempt( - plan: OfflineReadRequestPlan, - errorHandler: ErrorHandler, -): Observable | null> { +function resolveLocalAttempt(plan: OfflineReadRequestPlan, errorHandler: ErrorHandler): Observable | null> { return defer(() => from(plan.readLocal()).pipe( catchError((localError: unknown) => { @@ -136,10 +135,7 @@ function tryProjectLocal( ); } -function emitTaggedLocalResponse( - cached: AngularHttpResponse, - plan: OfflineReadRequestPlan, -): Observable> { +function emitTaggedLocalResponse(cached: AngularHttpResponse, plan: OfflineReadRequestPlan): Observable> { return projectReadResponse(cached.clone({ headers: cached.headers.set(OFFLINE_RESPONSE_HEADER, 'local') }), plan); } diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.ts index 6182dfc..4b8f327 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.ts @@ -47,6 +47,12 @@ import { } from './offline-repository'; import { OFFLINE_REPOSITORY_ATOMIC_MUTATION } from './offline-repository-concurrency'; import { OfflineStorageUnavailableError } from './offline-storage'; +import { + COMMUNITY_SQLITE_ENCRYPTED, + COMMUNITY_SQLITE_MODE, + COMMUNITY_SQLITE_READONLY, + COMMUNITY_SQLITE_VERSION, +} from './offline-community-sqlite-config'; /** Minimal native SQLite driver surface required by the offline repository. */ export interface CommunitySqliteDriver { @@ -118,7 +124,13 @@ export function createCommunitySqliteDriver(connection: CommunitySqliteConnectio if (!encryptionKey) throw new Error('Native offline storage requires a non-empty encryption key on first open'); await connection.setEncryptionSecret(encryptionKey); } - const value = await connection.createConnection(databaseName, true, 'secret', 1, false); + const value = await connection.createConnection( + databaseName, + COMMUNITY_SQLITE_ENCRYPTED, + COMMUNITY_SQLITE_MODE, + COMMUNITY_SQLITE_VERSION, + COMMUNITY_SQLITE_READONLY, + ); await value.open(); databases.set(databaseName, value); return { databaseId: databaseName }; diff --git a/projects/kit/offline/src/public-api.ts b/projects/kit/offline/src/public-api.ts index 9107fec..ffd6959 100644 --- a/projects/kit/offline/src/public-api.ts +++ b/projects/kit/offline/src/public-api.ts @@ -10,7 +10,14 @@ export * from './lib/offline-command-hooks'; export * from './lib/offline-auth-bridge'; export * from './lib/offline-coordinator.service'; export * from './lib/offline-kit-options'; +export * from './lib/offline-local-reset'; export * from './lib/offline-mutation-envelope'; +export { OfflineMutationPersistenceDisabledError } from './lib/offline-mutation-admission.service'; +export { + OfflineMutationPersistencePendingError, + OfflineMutationPersistenceRequiresOnlineError, + OfflineMutationPersistenceService, +} from './lib/offline-mutation-persistence.service'; export * from './lib/offline-network.service'; export * from './lib/offline-provider'; export * from './lib/offline-repository'; From 3d533f0a31a053b734c135b6993c6b86c1cf1c67 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Fri, 14 Aug 2026 16:39:02 +0900 Subject: [PATCH 2/2] fix(kit): allow unverified offline flush --- ...fline-mutation-persistence.service.spec.ts | 19 ++++++++++++++++++- .../offline-mutation-persistence.service.ts | 2 +- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-mutation-persistence.service.spec.ts b/projects/kit/offline/src/lib/offline-mutation-persistence.service.spec.ts index 03c5261..bdddade 100644 --- a/projects/kit/offline/src/lib/offline-mutation-persistence.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-mutation-persistence.service.spec.ts @@ -13,7 +13,7 @@ import { OfflineNetworkService } from './offline-network.service'; import { OfflineSyncService } from './offline-sync.service'; describe('OfflineMutationPersistenceService', () => { - const networkState = signal<'online' | 'offline'>('online'); + const networkState = signal<'online' | 'offline' | 'unverified'>('online'); const pendingCount = signal(0); const flush = vi.fn(async () => undefined); const handleError = vi.fn(); @@ -154,6 +154,23 @@ describe('OfflineMutationPersistenceService', () => { await expect(admission.run(async () => 'accepted')).resolves.toBe('accepted'); }); + it('attempts to flush pending commands while network reachability is unverified', async () => { + const saveEnabled = vi.fn(async () => undefined); + const { service } = setup(options(async () => true, saveEnabled)); + await service.initialize(); + networkState.set('unverified'); + pendingCount.set(1); + flush.mockImplementationOnce(async () => { + pendingCount.set(0); + }); + + await expect(service.setEnabled(false)).resolves.toBeUndefined(); + + expect(flush).toHaveBeenCalledOnce(); + expect(saveEnabled).toHaveBeenCalledExactlyOnceWith(false); + expect(service.enabled()).toBe(false); + }); + it('does not persist OFF when commands remain after flush', async () => { const saveEnabled = vi.fn(async () => undefined); const { service } = setup(options(async () => true, saveEnabled)); diff --git a/projects/kit/offline/src/lib/offline-mutation-persistence.service.ts b/projects/kit/offline/src/lib/offline-mutation-persistence.service.ts index 422f30c..6f1194b 100644 --- a/projects/kit/offline/src/lib/offline-mutation-persistence.service.ts +++ b/projects/kit/offline/src/lib/offline-mutation-persistence.service.ts @@ -153,7 +153,7 @@ export class OfflineMutationPersistenceService { if (revision !== this.#transitionRevision) return; const pendingBeforeFlush = this.#sync.pendingCount(); if (pendingBeforeFlush > 0) { - if (this.#network.state() !== 'online') throw new OfflineMutationPersistenceRequiresOnlineError(); + if (this.#network.state() === 'offline') throw new OfflineMutationPersistenceRequiresOnlineError(); await this.#sync.flush(); } if (revision !== this.#transitionRevision) return;