diff --git a/angular.json b/angular.json index ba6229e..bff15f1 100644 --- a/angular.json +++ b/angular.json @@ -262,6 +262,7 @@ "../theme/src/**/*.spec.ts", "../review/src/**/*.spec.ts", "../live-update/src/**/*.spec.ts", + "../offline/src/**/*.spec.ts", "../auth-firebase/src/**/*.spec.ts", "../auth-firebase/social/src/**/*.spec.ts" ] @@ -271,6 +272,12 @@ "watch": false } } + }, + "lint": { + "builder": "@angular-eslint/builder:lint", + "options": { + "lintFilePatterns": ["projects/kit/**/*.ts", "projects/kit/**/*.html"] + } } } } diff --git a/package-lock.json b/package-lock.json index bccaa10..81c826c 100644 --- a/package-lock.json +++ b/package-lock.json @@ -76,14 +76,14 @@ }, "dist/photo-editor": { "name": "@rdlabo/ionic-angular-photo-editor", - "version": "22.0.0", + "version": "21.4.0", "dependencies": { "tslib": "^2.3.0" }, "peerDependencies": { - "@angular/cdk": "^22.0.0", - "@angular/common": "^22.0.0", - "@angular/core": "^22.0.0", + "@angular/cdk": "^21.0.0", + "@angular/common": "^21.0.0", + "@angular/core": "^21.0.0", "@capacitor/camera": ">=6.0.0 <9.0.0", "@capacitor/core": ">=6.0.0 <9.0.0", "@ionic/angular": "^8.0.0", @@ -93,26 +93,26 @@ }, "dist/scroll-header": { "name": "@rdlabo/ionic-angular-scroll-header", - "version": "22.0.0", + "version": "21.4.0", "dependencies": { "tslib": "^2.3.0" }, "peerDependencies": { - "@angular/cdk": "^22.0.0", - "@angular/common": "^22.0.0", - "@angular/core": "^22.0.0" + "@angular/cdk": "^21.0.0", + "@angular/common": "^21.0.0", + "@angular/core": "^21.0.0" } }, "dist/scroll-strategies": { "name": "@rdlabo/ngx-cdk-scroll-strategies", - "version": "22.0.0", + "version": "21.4.0", "dependencies": { "tslib": "^2.3.0" }, "peerDependencies": { - "@angular/cdk": "^22.0.0", - "@angular/common": "^22.0.0", - "@angular/core": "^22.0.0" + "@angular/cdk": "^21.0.0", + "@angular/common": "^21.0.0", + "@angular/core": "^21.0.0" } }, "node_modules/@algolia/abtesting": { diff --git a/projects/kit/README.md b/projects/kit/README.md index eb66455..41268af 100644 --- a/projects/kit/README.md +++ b/projects/kit/README.md @@ -330,6 +330,129 @@ A fleet-canonical HTTP interceptor with: - Configurable bypass (CDN, S3, external URLs) - **Safe retry**: only idempotent methods (`GET`/`HEAD`/`OPTIONS`, or a request with an `Idempotency-Key`) are retried, and only on a transient status `[0, 408, 429, 502, 503, 504]`, up to 2 times with a short jittered backoff (honoring `Retry-After`). **Writes are never auto-retried** (no duplicate saves). - **Offline fast-fail**: when the device is offline the interceptor stops retrying immediately and hands off to `offlineFallback` instead of waiting out the backoff. + +### Scoped local replica and outbox (`@rdlabo/ionic-angular-kit/offline`) + +The optional `offline` entry point provides a user/group-scoped local replica, durable outbox, authenticated +session boundary, cursor-based delta pull, aggregate-ordered replay, optimistic updates, retry classification, and a +read-only request-policy interceptor. Applications provide URL/DTO read policies, a replica puller, and a command +executor through `provideOffline(...)`. +Mutations are queued explicitly with `OfflineSyncService.enqueue`, not through HTTP interceptor policy. +Web storage uses Ionic Storage; iOS and Android use encrypted Capawesome SQLite. Importing either the +primary entry point or `/offline` does not pull the private plugin into existing applications. + +The offline interceptor observes real transport responses to update API reachability. For matched `GET` +requests only, a transport failure with `status=0` may return a local replica response tagged +`X-Offline-Response: local`. `POST` and other write methods always go to transport unchanged; outbox replay +requests bypass policy with `OFFLINE_BYPASS` while still using the same transport observation. + +Native applications install the Insiders package in the application, then pass its `Sqlite` export +and an encryption key loaded from secure device storage to the kit: + +```bash +npm install @capawesome-team/capacitor-sqlite +``` + +```ts +provideOffline({ + // ...product policies, puller, and executor + sqlitePlugin: Sqlite, + encryptionKey: async () => { + const { value } = await securePreferences.get({ key: 'offline-database-key' }); + if (!value) throw new Error('offline-database-key is missing'); + return value; + }, +}); +``` + +SQLite entities use an immutable client-generated UUID as `localId` and keep the server's +`AUTO_INCREMENT` id separately as nullable `serverId`. The outbox references only `aggregateLocalId`. +Immediately before each send, the executor receives the latest `{ localId, serverId }` resolved from +SQLite; a successful create adds `serverId` without replacing `localId`. Entity projection and outbox +append/removal are committed in one local transaction. + +Each synchronization cycle pulls authoritative server deltas before replaying the outbox. Every page carries the +replica schema version/hash and advances a durable user/group cursor in the same transaction as its rows. A schema +mismatch, malformed row, or non-advancing cursor rejects synchronization without advancing that cursor. If a remote +revision changed while a local command is pending, the optimistic row remains visible and both row and command move +to `conflict`; the new server value is retained as the confirmed baseline. + +The command adapter must send `commandId` as the server-side idempotency key. The server persists that key with the +mutation and returns all keys represented by a delta row as `acknowledgedCommandIds`. This correlation is required: +if the server commits a create/update/delete but its HTTP acknowledgement is lost, the next pull reconciles the +server result into the original `localId`, removes the acknowledged outbox prefix, and rebases later commands without +creating a second local identity. + +Versioned replica schemas lock web and native storage. Web metadata stores +`replicaSchemaVersion` and `replicaSchemaHash`; native stores the same pair in +`offline_replica_schema_metadata`. Bump `version` for every intentional shape change and supply a +complete one-step migration chain. Native runs each step's SQL `statements`; web runs +`migrateWebRow`, which receives only `{ sourceKey, values, confirmedValues }` and may return the same +shape or `null` to delete a row. Identity and sync metadata (`localId`, `serverId`, scope, revision, +`syncState`) stay outside the callback. The bundle fingerprint hashes `version`, entity layouts, and +migration `fromVersion`/`statements` — never function bodies. + +```typescript +import { + defineOfflineReplicaSchema, + defineReplicaEntity, + provideOffline, + serverId, + text, +} from '@rdlabo/ionic-angular-kit/offline'; + +// This is the Hono package's existing `typeof items.$inferSelect` export. +import type { Items as ItemSelect } from '@product/hono/db/schema'; + +const itemEntityV2 = defineReplicaEntity()({ + table: 'items', + sourceKey: 'items', + scope: 'group', + fields: { + id: serverId(), + title: text(), + subtitle: text(), + }, +}); + +const replicaSchema = defineOfflineReplicaSchema({ + version: 2, + entities: [itemEntityV2], + migrations: [ + { + fromVersion: 1, + statements: ['ALTER TABLE items ADD COLUMN subtitle TEXT NOT NULL DEFAULT ""'], + migrateWebRow: (row) => ({ + sourceKey: row.sourceKey, + values: { ...row.values, subtitle: '' }, + confirmedValues: + row.confirmedValues === null ? null : { ...row.confirmedValues, subtitle: '' }, + }), + }, + ], +}); + +provideOffline({ + replicaSchema, + replicaPuller: ProductReplicaPuller, + commandExecutor: ProductCommandExecutor, + // ...request policies, sqlitePlugin, encryptionKey +}); +``` + +The schema definition must map every `ItemSelect` key exactly once as a SQLite column, `serverId()`, or +`ignored(reason)`, with exactly one `serverId()` per replicated entity. Nullable Hono columns require +`nullable(...)`; non-null columns reject it. Therefore adding, +removing, or changing nullability of a Drizzle column breaks the app build until its replica mapping is updated. +At runtime, `values` contains only the mapped column projection; `localId` and `serverId` remain dedicated replica +fields and ignored server fields are never persisted. + +Encrypted native builds also require the plugin's SQLCipher platform setup: enable +`capawesomeCapacitorSqliteIncludeSqlcipher = true` on Android; select the `SQLCipher` pod when using +CocoaPods, or enable the `SQLCipher` package trait when using Swift Package Manager on iOS. Follow +the [Capawesome SQLite installation guide](https://capawesome.io/docs/plugins/sqlite/#installation) +for the exact native configuration and export-compliance notes. + - **Status classification**: `0`→`onNetworkError` (connected only), `429`→`onRateLimited`, `502/503/504`→`onServerBusy`, `400/422/500`+message→`onServerError`, `401`→`onUnauthorized`, `403`→`onForbidden`. Other statuses (e.g. `404`) are left to the caller. - **Universal 60s timeout** — every request fails with a synthetic (retryable) `408` if it hangs for 60s. Deliberately generous (catches a dead server without cutting off a large upload / AI generation; `timeout({ each })` resets per emission, so streaming is unaffected). Not configurable — one fleet-wide behavior. - **Optional `treatAsError(response)`** — reject a 2xx (e.g. `204`/`206`) as an error when a backend uses it to signal a condition. The one genuinely app-specific bit (some apps receive a normal `204`), kept optional so class interceptors with a 2xx-as-error convention can migrate to `provideKitHttp`. diff --git a/projects/kit/offline/ng-package.json b/projects/kit/offline/ng-package.json new file mode 100644 index 0000000..d0a2dcd --- /dev/null +++ b/projects/kit/offline/ng-package.json @@ -0,0 +1,6 @@ +{ + "$schema": "../../../node_modules/ng-packagr/ng-package.schema.json", + "lib": { + "entryFile": "src/public-api.ts" + } +} diff --git a/projects/kit/offline/src/lib/offline-command-executor.ts b/projects/kit/offline/src/lib/offline-command-executor.ts new file mode 100644 index 0000000..2683617 --- /dev/null +++ b/projects/kit/offline/src/lib/offline-command-executor.ts @@ -0,0 +1,45 @@ +import { InjectionToken } from '@angular/core'; +import type { OfflineCommand, OfflineScope } from './offline-repository'; + +/** Server acknowledgement used to reconcile one optimistic local mutation. */ +export interface OfflineCommandResult { + /** AUTO_INCREMENT id returned by a successful create. */ + serverId?: number; + serverRevision?: string | number; + /** Full server-confirmed domain values after applying the mutation. */ + confirmedValues?: unknown; + /** Removes the local replica row after a confirmed server delete. */ + removeReplica?: boolean; + response?: unknown; +} + +/** Target ids resolved from the local replica immediately before transport. */ +export interface OfflineCommandTarget { + localId: string; + serverId: number | null; +} + +/** 不透明なoperationを製品APIへ送信し、local replicaへ投影するadapter。 */ +/** Product adapter that sends commands and projects acknowledgements into entities. */ +export interface OfflineCommandExecutor { + /** Sends the command using `command.commandId` as its durable server-side idempotency key. */ + execute(command: OfflineCommand, target: OfflineCommandTarget): Promise; + withServerRevision(command: OfflineCommand, revision: string | number): OfflineCommand; +} + +/** DI token for the product-specific command transport adapter. */ +export const OFFLINE_COMMAND_EXECUTOR = new InjectionToken('OFFLINE_COMMAND_EXECUTOR'); + +/** Authenticated user and group scopes currently eligible for synchronization. */ +export interface OfflineSyncSession { + userId: number; + scopes: OfflineScope[]; +} + +/** Product adapter that exposes the currently authenticated synchronization session. */ +export interface OfflineSyncContext { + getSession(): Promise; +} + +/** DI token for authenticated synchronization context. */ +export const OFFLINE_SYNC_CONTEXT = new InjectionToken('OFFLINE_SYNC_CONTEXT'); diff --git a/projects/kit/offline/src/lib/offline-command-hooks.ts b/projects/kit/offline/src/lib/offline-command-hooks.ts new file mode 100644 index 0000000..c9928eb --- /dev/null +++ b/projects/kit/offline/src/lib/offline-command-hooks.ts @@ -0,0 +1,18 @@ +import { InjectionToken } from '@angular/core'; +import type { OfflineCommand } from './offline-repository'; + +/** Optional product hooks for entity projection and command cleanup. */ +export interface OfflineCommandHooks { + entityType(command: Pick): string; + onCommandRemoved?(command: OfflineCommand): Promise; +} + +export const DEFAULT_OFFLINE_COMMAND_HOOKS: OfflineCommandHooks = { + entityType: (command) => command.aggregateType, +}; + +/** DI token for optional product-specific synchronization hooks. */ +export const OFFLINE_COMMAND_HOOKS = new InjectionToken('OFFLINE_COMMAND_HOOKS', { + providedIn: 'root', + factory: () => DEFAULT_OFFLINE_COMMAND_HOOKS, +}); diff --git a/projects/kit/offline/src/lib/offline-coordinator.service.ts b/projects/kit/offline/src/lib/offline-coordinator.service.ts new file mode 100644 index 0000000..8e75946 --- /dev/null +++ b/projects/kit/offline/src/lib/offline-coordinator.service.ts @@ -0,0 +1,53 @@ +import { inject, Injectable } from '@angular/core'; +import { OfflineNetworkService } from './offline-network.service'; +import { OFFLINE_REPOSITORY } from './offline-repository'; +import { OfflineSessionService } from './offline-session.service'; +import { OfflineSyncService } from './offline-sync.service'; + +/** User choice when logout encounters unconfirmed local mutations. */ +export type OfflineLogoutAction = 'sync' | 'discard' | 'cancel'; + +/** Coordinates local persistence, session boundaries, network state, and outbox synchronization. */ +@Injectable({ providedIn: 'root' }) +export class OfflineCoordinatorService { + readonly #repository = inject(OFFLINE_REPOSITORY); + readonly #network = inject(OfflineNetworkService); + readonly #sync = inject(OfflineSyncService); + readonly #session = inject(OfflineSessionService); + + readonly networkState = this.#network.state; + readonly syncState = this.#sync.syncState; + readonly pendingCount = this.#sync.pendingCount; + readonly conflicts = this.#sync.conflicts; + + async initialize(): Promise { + await Promise.all([this.#repository.initialize(), this.#network.initialize()]); + await this.#session.initialize(); + await this.#sync.initialize(); + } + + async activateSession(userId: number, scopeIds: readonly number[], authSubject: string | null): Promise { + await this.#sync.resetSession(); + await this.#session.activateSession(userId, scopeIds, authSubject); + await this.#sync.refreshSession(); + } + + async clearActiveSession(): Promise { + await this.#sync.resetSession(); + await this.#session.clearActiveSession(); + } + + async prepareLogout(action: OfflineLogoutAction): Promise { + if (action === 'cancel') return false; + if (action === 'discard') { + await this.#sync.discardAllPending(); + return true; + } + await this.#sync.flush(); + return this.#sync.pendingCount() === 0; + } + + flush(): Promise { + return this.#sync.flush(); + } +} diff --git a/projects/kit/offline/src/lib/offline-kit-options.ts b/projects/kit/offline/src/lib/offline-kit-options.ts new file mode 100644 index 0000000..b7484f5 --- /dev/null +++ b/projects/kit/offline/src/lib/offline-kit-options.ts @@ -0,0 +1,15 @@ +import { InjectionToken } from '@angular/core'; +import type { OfflineReplicaSchemaBundle } from './offline-replica-schema'; + +/** Product-independent native offline persistence settings. */ +export interface OfflineKitOptions { + /** Encrypted SQLite database name used on iOS and Android. */ + databaseName: string; + /** Resolves the native database key from secure device storage. Required on iOS and Android. */ + encryptionKey?: () => Promise; + /** Versioned product replica schema applied to native SQLite during initialization. */ + replicaSchema: OfflineReplicaSchemaBundle; +} + +/** DI token for product-independent offline persistence settings. */ +export const OFFLINE_KIT_OPTIONS = new InjectionToken('OFFLINE_KIT_OPTIONS'); diff --git a/projects/kit/offline/src/lib/offline-network.service.ts b/projects/kit/offline/src/lib/offline-network.service.ts new file mode 100644 index 0000000..6238fd6 --- /dev/null +++ b/projects/kit/offline/src/lib/offline-network.service.ts @@ -0,0 +1,54 @@ +import { computed, Injectable, signal } from '@angular/core'; +import { App } from '@capacitor/app'; +import type { PluginListenerHandle } from '@capacitor/core'; +import { Network } from '@capacitor/network'; + +export type OfflineNetworkState = 'online' | 'offline' | 'unverified'; + +/** transport不能(status=0)だけをlocal replica fallback対象にし、HTTPエラーは隠さない。 */ +export function isOfflineFallbackError(error: unknown): boolean { + return typeof error === 'object' && error !== null && (error as { status?: unknown }).status === 0; +} + +/** Combines operating-system connectivity with observed API reachability. */ +@Injectable({ providedIn: 'root' }) +export class OfflineNetworkService { + readonly #osConnected = signal(null); + readonly #apiReachable = signal(null); + readonly #listeners: PluginListenerHandle[] = []; + #initialized = false; + + readonly state = computed(() => { + if (this.#osConnected() === false || this.#apiReachable() === false) return 'offline'; + if (this.#osConnected() === true && this.#apiReachable() === true) return 'online'; + return 'unverified'; + }); + readonly connected = computed(() => this.state() !== 'offline'); + + async initialize(): Promise { + if (this.#initialized) return; + this.#initialized = true; + this.#osConnected.set((await Network.getStatus()).connected); + this.#listeners.push( + await Network.addListener('networkStatusChange', ({ connected }) => { + this.#osConnected.set(connected); + this.#apiReachable.set(connected ? null : false); + }), + await App.addListener('appStateChange', ({ isActive }) => { + if (isActive) void this.#refreshOsStatus(); + }), + ); + } + + markApiSuccess(): void { + this.#apiReachable.set(true); + } + + markApiFailure(): void { + this.#apiReachable.set(false); + } + + async #refreshOsStatus(): Promise { + this.#osConnected.set((await Network.getStatus()).connected); + } +} diff --git a/projects/kit/offline/src/lib/offline-provider.ts b/projects/kit/offline/src/lib/offline-provider.ts new file mode 100644 index 0000000..10ea7b9 --- /dev/null +++ b/projects/kit/offline/src/lib/offline-provider.ts @@ -0,0 +1,68 @@ +import type { EnvironmentProviders, Provider, Type } from '@angular/core'; +import { inject, makeEnvironmentProviders, provideAppInitializer } from '@angular/core'; +import { Capacitor } from '@capacitor/core'; +import type { OfflineCommandExecutor } from './offline-command-executor'; +import { OFFLINE_COMMAND_EXECUTOR, OFFLINE_SYNC_CONTEXT } from './offline-command-executor'; +import type { OfflineCommandHooks } from './offline-command-hooks'; +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 { IonicOfflineRepository, OFFLINE_REPOSITORY, selectOfflineRepository } from './offline-repository'; +import type { OfflineRequestPolicy } from './offline-request-policy'; +import { provideOfflineRequestPolicy } from './offline-request-policy'; +import type { OfflineReplicaPuller } from './offline-replica-puller'; +import { OFFLINE_REPLICA_PULLER } from './offline-replica-puller'; +import { OfflineSessionService } from './offline-session.service'; +import { CAPAWESOME_SQLITE, type CapawesomeSqlitePlugin, SqliteOfflineRepository } from './sqlite-offline-repository'; + +/** Configuration for the standard offline repository, outbox, and request-policy runtime. */ +export interface ProvideOfflineOptions extends OfflineKitOptions { + /** Product adapter that sends opaque commands to its API. */ + commandExecutor: Type; + /** Product transport for explicit cursor-based server delta pulls. */ + replicaPuller: Type; + /** Product policies that map URLs and DTOs to generic replica/outbox operations. */ + requestPolicies: readonly Type[]; + /** Optional product hooks for entity projection and command cleanup. */ + commandHooks?: Type; + /** Optional additional providers required by product adapters. */ + providers?: readonly Provider[]; + /** Capawesome `Sqlite` plugin. Required only when this runtime is selected on iOS or Android. */ + sqlitePlugin?: CapawesomeSqlitePlugin; +} + +/** + * Provide the standard scoped offline runtime. + * + * @remarks + * Web uses Ionic Storage. Native iOS/Android uses encrypted Capawesome SQLite. The application owns + * URL/DTO policy and command execution; the kit owns persistence, ordering, retries, and session + * isolation. + */ +export function provideOffline(options: ProvideOfflineOptions): EnvironmentProviders { + return makeEnvironmentProviders([ + options.commandExecutor, + options.replicaPuller, + { + provide: OFFLINE_KIT_OPTIONS, + useValue: { + databaseName: options.databaseName, + encryptionKey: options.encryptionKey, + replicaSchema: options.replicaSchema, + }, + }, + { provide: CAPAWESOME_SQLITE, useValue: options.sqlitePlugin ?? null }, + { + provide: OFFLINE_REPOSITORY, + useFactory: () => selectOfflineRepository(Capacitor.getPlatform(), inject(IonicOfflineRepository), inject(SqliteOfflineRepository)), + }, + { provide: OFFLINE_SYNC_CONTEXT, useExisting: OfflineSessionService }, + { provide: OFFLINE_COMMAND_EXECUTOR, useExisting: options.commandExecutor }, + { provide: OFFLINE_REPLICA_PULLER, useExisting: options.replicaPuller }, + ...(options.commandHooks ? [options.commandHooks, { provide: OFFLINE_COMMAND_HOOKS, useExisting: options.commandHooks }] : []), + ...options.requestPolicies.flatMap((policy) => provideOfflineRequestPolicy(policy)), + ...(options.providers ?? []), + provideAppInitializer(() => inject(OfflineCoordinatorService).initialize()), + ]); +} diff --git a/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts new file mode 100644 index 0000000..01f358a --- /dev/null +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts @@ -0,0 +1,991 @@ +/* eslint-disable @typescript-eslint/consistent-type-definitions */ +import { TestBed } from '@angular/core/testing'; +import { KitStorageService } from '@rdlabo/ionic-angular-kit'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { OFFLINE_COMMAND_EXECUTOR } from './offline-command-executor'; +import { OFFLINE_COMMAND_HOOKS } from './offline-command-hooks'; +import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; +import { + OFFLINE_REPLICA_PULLER, + type OfflineReplicaChange, + type OfflineReplicaPullPage, + type OfflineReplicaPullRequest, +} from './offline-replica-puller'; +import { OfflineReplicaPullService } from './offline-replica-pull.service'; +import { defineOfflineReplicaSchema, defineReplicaEntity, serverId, sha256OfflineReplicaSchema, text } from './offline-replica-schema'; +import { + IonicOfflineRepository, + OFFLINE_REPOSITORY, + OFFLINE_SCHEMA_VERSION, + type OfflineCommand, + type OfflineReplicaRow, + type OfflineRepository, + type OfflineScope, +} from './offline-repository'; + +type TestItemSelect = { id: number; title: string }; + +const testItemEntity = defineReplicaEntity()({ + table: 'test_items', + sourceKey: 'test_items', + scope: 'user', + fields: { + id: serverId(), + title: text(), + }, +}); + +const replicaSchema = defineOfflineReplicaSchema({ + version: 1, + entities: [testItemEntity], + migrations: [], +}); + +const scope: OfflineScope = { userId: 1, groupId: 10 }; + +class MemoryStorage { + readonly values = new Map(); + get(key: string): Promise { + return Promise.resolve((this.values.get(key) as T | undefined) ?? null); + } + set(key: string, value: T): Promise { + this.values.set(key, structuredClone(value)); + return Promise.resolve(value); + } + remove(key: string): Promise { + this.values.delete(key); + return Promise.resolve(); + } + keys(): Promise { + return Promise.resolve([...this.values.keys()]); + } +} + +function itemChange( + serverIdValue: number, + title: string, + options: Partial> = {}, +): OfflineReplicaChange { + return { + sourceKey: 'test_items', + serverId: serverIdValue, + serverRevision: options.serverRevision ?? 1, + acknowledgedCommandIds: options.acknowledgedCommandIds ?? [], + values: options.deleted ? null : (options.values ?? { id: serverIdValue, title }), + deleted: options.deleted ?? false, + }; +} + +describe('OfflineReplicaPullService', () => { + let service: OfflineReplicaPullService; + let repository: OfflineRepository; + let storage: MemoryStorage; + let schemaHash: string; + let pull: ReturnType Promise>>; + + function page( + changes: readonly OfflineReplicaChange[], + options: { + nextCursor?: string; + hasMore?: boolean; + schemaVersion?: number; + schemaHash?: string; + } = {}, + ): OfflineReplicaPullPage { + return { + schemaVersion: options.schemaVersion ?? replicaSchema.version, + schemaHash: options.schemaHash ?? schemaHash, + changes, + nextCursor: options.nextCursor ?? 'cursor-v1', + hasMore: options.hasMore ?? false, + }; + } + + async function expectPullRejectsPreservingCursor(setup: () => void, message: string | RegExp): Promise { + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); + setup(); + await expect(service.pull(scope)).rejects.toThrow(message); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); + expect(await repository.getReplicaRows(scope, 'test_items')).toEqual([]); + } + + async function seedReplicaMetadata(): Promise { + schemaHash = await sha256OfflineReplicaSchema(replicaSchema); + storage.values.set('offline:metadata', { + schemaVersion: OFFLINE_SCHEMA_VERSION, + lastUserId: null, + replicaSchemaVersion: replicaSchema.version, + replicaSchemaHash: schemaHash, + }); + storage.values.set('offline:replica:rows', {}); + storage.values.set('offline:outbox:commands', {}); + storage.values.set('offline:replica:cursors', {}); + } + + function configureTestBed(): void { + TestBed.resetTestingModule(); + TestBed.configureTestingModule({ + providers: [ + OfflineReplicaPullService, + IonicOfflineRepository, + { provide: KitStorageService, useValue: storage }, + { provide: OFFLINE_KIT_OPTIONS, useValue: { databaseName: 'test-offline', replicaSchema } }, + { provide: OFFLINE_REPOSITORY, useExisting: IonicOfflineRepository }, + { provide: OFFLINE_REPLICA_PULLER, useValue: { pull } }, + { + provide: OFFLINE_COMMAND_HOOKS, + useValue: { entityType: (command: Pick) => command.aggregateType }, + }, + { + provide: OFFLINE_COMMAND_EXECUTOR, + useValue: { + execute: vi.fn(), + withServerRevision: (command: OfflineCommand, revision: string | number) => ({ + ...command, + baseRevision: revision, + }), + }, + }, + ], + }); + repository = TestBed.inject(OFFLINE_REPOSITORY); + service = TestBed.inject(OfflineReplicaPullService); + } + + beforeEach(async () => { + storage = new MemoryStorage(); + pull = vi.fn(async () => page([])); + await seedReplicaMetadata(); + configureTestBed(); + await repository.initialize(); + }); + + it('initial empty cursor requestをpullerへ送る', async () => { + pull.mockResolvedValueOnce(page([itemChange(42, 'Created')], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + expect(pull).toHaveBeenCalledOnce(); + expect(pull.mock.calls[0]?.[0]).toEqual({ + scope, + cursor: '', + schemaVersion: replicaSchema.version, + schemaHash, + }); + }); + + it('exact schema version/hash handshakeを要求し、一致ページだけ受理する', async () => { + pull.mockResolvedValueOnce(page([itemChange(42, 'Created')], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + expect(pull.mock.calls[0]?.[0].schemaVersion).toBe(1); + expect(pull.mock.calls[0]?.[0].schemaHash).toBe(schemaHash); + }); + + it('multi-page cursor progressionでstored cursorをページングする', async () => { + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); + pull + .mockResolvedValueOnce(page([itemChange(42, 'Page 1')], { nextCursor: 'cursor-v1', hasMore: true })) + .mockResolvedValueOnce(page([itemChange(43, 'Page 2')], { nextCursor: 'cursor-v2', hasMore: false })); + + await service.pull(scope); + + expect(pull.mock.calls.map(([request]) => request.cursor)).toEqual(['cursor-v0', 'cursor-v1']); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v2' }); + await expect(repository.getReplicaRowByServerId(scope, 'test_items', 42)).resolves.toMatchObject({ + confirmedValues: { title: 'Page 1' }, + }); + await expect(repository.getReplicaRowByServerId(scope, 'test_items', 43)).resolves.toMatchObject({ + confirmedValues: { title: 'Page 2' }, + }); + }); + + it('row更新とcursor更新を同一transactReplica呼び出しで原子的に書く', async () => { + const transactReplica = vi.spyOn(repository, 'transactReplica'); + pull.mockResolvedValueOnce(page([itemChange(42, 'Created')], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + expect(transactReplica).toHaveBeenCalledOnce(); + expect(transactReplica.mock.calls[0]?.[0]).toMatchObject({ + putRows: [expect.objectContaining({ serverId: 42, confirmedValues: { title: 'Created' } })], + putCursors: [{ ...scope, cursor: 'cursor-v1' }], + }); + }); + + it('new remote rowにlocal UUIDとserver IDを割り当てる', async () => { + const randomUuid = vi.spyOn(crypto, 'randomUUID').mockReturnValue('019d0000-0000-7000-8000-000000000001'); + pull.mockResolvedValueOnce(page([itemChange(42, 'Created')], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + await expect(repository.getReplicaRow(scope, 'test_items', '019d0000-0000-7000-8000-000000000001')).resolves.toMatchObject({ + localId: '019d0000-0000-7000-8000-000000000001', + serverId: 42, + sourceKey: 'test_items', + syncState: 'confirmed', + values: { title: 'Created' }, + confirmedValues: { title: 'Created' }, + }); + randomUuid.mockRestore(); + }); + + it('existing remote rowをupdateする', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + localId: '019d-existing', + serverId: 42, + values: { id: 42, title: 'Old' }, + confirmedValues: { id: 42, title: 'Old' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Updated', { serverRevision: 2 })], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + await expect(repository.getReplicaRow(scope, 'test_items', '019d-existing')).resolves.toMatchObject({ + localId: '019d-existing', + serverId: 42, + serverRevision: 2, + values: { title: 'Updated' }, + confirmedValues: { title: 'Updated' }, + syncState: 'confirmed', + }); + }); + + it('pending commandが無いdeleteはreplica rowを削除する', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + localId: '019d-delete', + serverId: 42, + values: { id: 42, title: 'Gone' }, + confirmedValues: { id: 42, title: 'Gone' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Gone', { deleted: true, serverRevision: 2 })], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + expect(await repository.getReplicaRow(scope, 'test_items', '019d-delete')).toBeNull(); + expect(await repository.getReplicaRowByServerId(scope, 'test_items', 42)).toBeNull(); + }); + + it('duplicate changeはlast-winsでcollapseする', async () => { + pull.mockResolvedValueOnce( + page( + [ + itemChange(42, 'First', { serverRevision: 1 }), + itemChange(42, 'Second', { serverRevision: 2 }), + itemChange(42, 'Third', { serverRevision: 3 }), + ], + { nextCursor: 'cursor-v1' }, + ), + ); + + await service.pull(scope); + + const rows = await repository.getReplicaRows(scope, 'test_items'); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + confirmedValues: { title: 'Third' }, + serverRevision: 3, + }); + }); + + it('invalid valuesはrejectしcursorを進めない', async () => { + await expectPullRejectsPreservingCursor( + () => pull.mockResolvedValueOnce(page([itemChange(42, 'Broken', { values: { id: 42 } })], { nextCursor: 'cursor-v1' })), + 'Replica row is missing required source key "title".', + ); + }); + + describe('pull page boundary validation', () => { + it('malformed nextCursorはrejectしcursorを進めない', async () => { + await expectPullRejectsPreservingCursor( + () => + pull.mockResolvedValueOnce({ + ...page([itemChange(42, 'Created')], { nextCursor: 'cursor-v1' }), + nextCursor: 1 as unknown as string, + }), + 'Offline replica pull page nextCursor must be a string.', + ); + }); + + it('malformed hasMoreはrejectしcursorを進めない', async () => { + await expectPullRejectsPreservingCursor( + () => + pull.mockResolvedValueOnce({ + ...page([itemChange(42, 'Created')], { nextCursor: 'cursor-v1' }), + hasMore: 'yes' as unknown as boolean, + }), + 'Offline replica pull page hasMore must be a boolean.', + ); + }); + + it('malformed changesはrejectしcursorを進めない', async () => { + await expectPullRejectsPreservingCursor( + () => + pull.mockResolvedValueOnce({ + ...page([itemChange(42, 'Created')], { nextCursor: 'cursor-v1' }), + changes: null as unknown as OfflineReplicaChange[], + }), + 'Offline replica pull page changes must be an array.', + ); + }); + + it('non-positive serverIdはrejectしcursorを進めない', async () => { + await expectPullRejectsPreservingCursor( + () => pull.mockResolvedValueOnce(page([{ ...itemChange(42, 'Created'), serverId: 0 }], { nextCursor: 'cursor-v1' })), + 'Offline replica pull page changes[0].serverId must be a positive integer.', + ); + }); + + it('non-integer serverIdはrejectしcursorを進めない', async () => { + await expectPullRejectsPreservingCursor( + () => pull.mockResolvedValueOnce(page([{ ...itemChange(42, 'Created'), serverId: 42.5 }], { nextCursor: 'cursor-v1' })), + 'Offline replica pull page changes[0].serverId must be a positive integer.', + ); + }); + + it('invalid serverRevision typeはrejectしcursorを進めない', async () => { + await expectPullRejectsPreservingCursor( + () => + pull.mockResolvedValueOnce( + page([{ ...itemChange(42, 'Created'), serverRevision: true as unknown as number }], { + nextCursor: 'cursor-v1', + }), + ), + 'Offline replica pull page changes[0].serverRevision must be a string or number.', + ); + }); + + it('deleted change with non-null valuesはrejectしcursorを進めない', async () => { + await expectPullRejectsPreservingCursor( + () => + pull.mockResolvedValueOnce( + page([{ ...itemChange(42, 'Gone', { deleted: true, serverRevision: 2 }), values: { id: 42, title: 'Gone' } }], { + nextCursor: 'cursor-v1', + }), + ), + 'Offline replica pull page changes[0] with deleted=true must have null values.', + ); + }); + }); + + it('unknown source keyはrejectしcursorを進めない', async () => { + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); + pull.mockResolvedValueOnce( + page( + [ + { + sourceKey: 'unknown_items', + serverId: 42, + serverRevision: 1, + acknowledgedCommandIds: [], + values: { id: 42, title: 'X' }, + deleted: false, + }, + ], + { nextCursor: 'cursor-v1' }, + ), + ); + + await expect(service.pull(scope)).rejects.toThrow('Unknown offline replica source key "unknown_items".'); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); + }); + + it('missing valuesはrejectしcursorを進めない', async () => { + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); + pull.mockResolvedValueOnce( + page([{ sourceKey: 'test_items', serverId: 42, serverRevision: 1, acknowledgedCommandIds: [], values: null, deleted: false }], { + nextCursor: 'cursor-v1', + }), + ); + + await expect(service.pull(scope)).rejects.toThrow('Offline replica change "test_items"/42 is missing values.'); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); + }); + + it('schema mismatchはrejectしcursorを進めない', async () => { + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Created')], { nextCursor: 'cursor-v1', schemaVersion: 99, schemaHash: 'deadbeef' })); + + await expect(service.pull(scope)).rejects.toThrow('Offline replica schema mismatch'); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); + expect(await repository.getReplicaRows(scope, 'test_items')).toEqual([]); + }); + + it('non-advancing cursorはrejectしcursorを進めない', async () => { + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Created')], { nextCursor: 'cursor-v0', hasMore: true })); + + await expect(service.pull(scope)).rejects.toThrow('Offline replica pull cursor did not advance'); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); + expect(await repository.getReplicaRows(scope, 'test_items')).toEqual([]); + }); + + it('pending optimistic rowはconfirmed baselineだけ更新しoptimistic valuesを保持する', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + localId: '019d-pending', + serverId: 42, + values: { id: 42, title: 'Optimistic draft' }, + confirmedValues: { id: 42, title: 'Confirmed baseline' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending', + }, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-pending', + aggregateType: 'test_items', + aggregateLocalId: '019d-pending', + operation: 'test_items.update', + payload: { title: 'Optimistic draft' }, + optimisticValue: { id: 42, title: 'Optimistic draft' }, + payloadHash: 'hash', + baseRevision: 2, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Server truth', { serverRevision: 2 })], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + await expect(repository.getReplicaRow(scope, 'test_items', '019d-pending')).resolves.toMatchObject({ + values: { title: 'Optimistic draft' }, + confirmedValues: { title: 'Server truth' }, + serverRevision: 2, + syncState: 'pending', + }); + await expect(repository.getCommands(scope)).resolves.toEqual([expect.objectContaining({ commandId: 'cmd-pending', state: 'pending' })]); + }); + + it('revision conflictはreplicaとcommandをconflictへ遷移する', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + localId: '019d-conflict', + serverId: 42, + values: { id: 42, title: 'Local edit' }, + confirmedValues: { id: 42, title: 'Old confirmed' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending', + }, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-conflict', + aggregateType: 'test_items', + aggregateLocalId: '019d-conflict', + operation: 'test_items.update', + payload: { title: 'Local edit' }, + optimisticValue: { id: 42, title: 'Local edit' }, + payloadHash: 'hash', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Remote truth', { serverRevision: 9 })], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + await expect(repository.getReplicaRow(scope, 'test_items', '019d-conflict')).resolves.toMatchObject({ + syncState: 'conflict', + confirmedValues: { title: 'Remote truth' }, + serverRevision: 9, + }); + await expect(repository.getCommands(scope)).resolves.toEqual([ + expect.objectContaining({ + commandId: 'cmd-conflict', + state: 'conflict', + lastErrorCode: 'remote_revision', + retryAt: null, + }), + ]); + }); + + it('remote tombstone conflictはpending commandをremote_deleted conflictへ遷移する', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + localId: '019d-tombstone', + serverId: 42, + values: { id: 42, title: 'Pending delete' }, + confirmedValues: { id: 42, title: 'Confirmed' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending', + }, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-tombstone', + aggregateType: 'test_items', + aggregateLocalId: '019d-tombstone', + operation: 'test_items.delete', + payload: {}, + optimisticValue: { id: 42, title: 'Pending delete' }, + payloadHash: 'hash', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Confirmed', { deleted: true, serverRevision: 2 })], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + await expect(repository.getReplicaRow(scope, 'test_items', '019d-tombstone')).resolves.toMatchObject({ + localId: '019d-tombstone', + syncState: 'conflict', + serverRevision: 2, + }); + await expect(repository.getCommands(scope)).resolves.toEqual([ + expect.objectContaining({ + commandId: 'cmd-tombstone', + state: 'conflict', + lastErrorCode: 'remote_deleted', + retryAt: null, + }), + ]); + expect(await repository.getReplicaRowByServerId(scope, 'test_items', 42)).not.toBeNull(); + }); + + describe('lost ACK correlation', () => { + async function seedPendingCreate(localId = '019d-create'): Promise { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + localId, + serverId: null, + values: { id: 0, title: 'Draft create' }, + confirmedValues: null, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + }, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-create', + aggregateType: 'test_items', + aggregateLocalId: localId, + operation: 'test_items.create', + payload: { title: 'Draft create' }, + optimisticValue: { id: 0, title: 'Draft create' }, + payloadHash: 'hash', + baseRevision: null, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + }); + } + + it('create lost ACKは既存localId行をreconcileしserverIdを割り当ててcommandを除去する', async () => { + await seedPendingCreate(); + pull.mockResolvedValueOnce( + page([itemChange(42, 'Created', { serverRevision: 1, acknowledgedCommandIds: ['cmd-create'] })], { nextCursor: 'cursor-v1' }), + ); + + await service.pull(scope); + + await expect(repository.getReplicaRow(scope, 'test_items', '019d-create')).resolves.toMatchObject({ + localId: '019d-create', + serverId: 42, + confirmedValues: { title: 'Created' }, + syncState: 'confirmed', + }); + expect(await repository.getCommands(scope)).toEqual([]); + expect(await repository.getReplicaRows(scope, 'test_items')).toHaveLength(1); + }); + + it('update lost ACKはprefix commandを除去しfollowing commandをrebaseする', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + localId: '019d-update', + serverId: 42, + values: { id: 42, title: 'Follow-up edit' }, + confirmedValues: { id: 42, title: 'Confirmed baseline' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending', + }, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-update-1', + aggregateType: 'test_items', + aggregateLocalId: '019d-update', + operation: 'test_items.update', + payload: { title: 'First edit' }, + optimisticValue: { id: 42, title: 'First edit' }, + payloadHash: 'hash-1', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + { + ...scope, + commandId: 'cmd-update-2', + aggregateType: 'test_items', + aggregateLocalId: '019d-update', + operation: 'test_items.update', + payload: { title: 'Follow-up edit' }, + optimisticValue: { id: 42, title: 'Follow-up edit' }, + payloadHash: 'hash-2', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 2, + lastErrorCode: null, + }, + ], + }); + pull.mockResolvedValueOnce( + page([itemChange(42, 'First edit applied', { serverRevision: 2, acknowledgedCommandIds: ['cmd-update-1'] })], { + nextCursor: 'cursor-v1', + }), + ); + + await service.pull(scope); + + await expect(repository.getReplicaRow(scope, 'test_items', '019d-update')).resolves.toMatchObject({ + localId: '019d-update', + serverId: 42, + values: { title: 'Follow-up edit' }, + confirmedValues: { title: 'First edit applied' }, + serverRevision: 2, + syncState: 'pending', + }); + expect(await repository.getCommands(scope)).toEqual([ + expect.objectContaining({ commandId: 'cmd-update-2', baseRevision: 2, state: 'pending' }), + ]); + }); + + it('delete lost ACKはfollowing commandが無ければ行を削除する', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + localId: '019d-delete-ack', + serverId: 42, + values: { id: 42, title: 'Pending delete' }, + confirmedValues: { id: 42, title: 'Confirmed' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending', + }, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-delete', + aggregateType: 'test_items', + aggregateLocalId: '019d-delete-ack', + operation: 'test_items.delete', + payload: {}, + optimisticValue: { id: 42, title: 'Pending delete' }, + payloadHash: 'hash', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + }); + pull.mockResolvedValueOnce( + page([itemChange(42, 'Confirmed', { deleted: true, serverRevision: 2, acknowledgedCommandIds: ['cmd-delete'] })], { + nextCursor: 'cursor-v1', + }), + ); + + await service.pull(scope); + + expect(await repository.getReplicaRow(scope, 'test_items', '019d-delete-ack')).toBeNull(); + expect(await repository.getReplicaRowByServerId(scope, 'test_items', 42)).toBeNull(); + expect(await repository.getCommands(scope)).toEqual([]); + }); + + it('duplicate deltaはacknowledgedCommandIdsをマージする', async () => { + await seedPendingCreate(); + pull.mockResolvedValueOnce( + page( + [ + itemChange(42, 'Partial', { serverRevision: 1, acknowledgedCommandIds: ['cmd-create'] }), + itemChange(42, 'Final', { serverRevision: 2, acknowledgedCommandIds: ['cmd-create'] }), + ], + { nextCursor: 'cursor-v1' }, + ), + ); + + await service.pull(scope); + + await expect(repository.getReplicaRow(scope, 'test_items', '019d-create')).resolves.toMatchObject({ + confirmedValues: { title: 'Final' }, + serverRevision: 2, + syncState: 'confirmed', + }); + expect(await repository.getCommands(scope)).toEqual([]); + }); + + it('skipped-prefix acknowledgementはrejectする', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + localId: '019d-skip', + serverId: 42, + values: { id: 42, title: 'Second edit' }, + confirmedValues: { id: 42, title: 'Baseline' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending', + }, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-first', + aggregateType: 'test_items', + aggregateLocalId: '019d-skip', + operation: 'test_items.update', + payload: { title: 'First edit' }, + optimisticValue: { id: 42, title: 'First edit' }, + payloadHash: 'hash-1', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + { + ...scope, + commandId: 'cmd-second', + aggregateType: 'test_items', + aggregateLocalId: '019d-skip', + operation: 'test_items.update', + payload: { title: 'Second edit' }, + optimisticValue: { id: 42, title: 'Second edit' }, + payloadHash: 'hash-2', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 2, + lastErrorCode: null, + }, + ], + }); + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); + pull.mockResolvedValueOnce( + page([itemChange(42, 'Only second', { serverRevision: 2, acknowledgedCommandIds: ['cmd-second'] })], { nextCursor: 'cursor-v1' }), + ); + + await expect(service.pull(scope)).rejects.toThrow('Replica acknowledgement skipped an earlier aggregate command.'); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); + expect(await repository.getCommands(scope)).toHaveLength(2); + }); + + it('server id collisionはrejectする', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + localId: '019d-local-a', + serverId: null, + values: { id: 0, title: 'Pending create A' }, + confirmedValues: null, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + }, + { + ...scope, + sourceKey: 'test_items', + localId: '019d-local-b', + serverId: 99, + values: { id: 99, title: 'Existing remote' }, + confirmedValues: { id: 99, title: 'Existing remote' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-create-a', + aggregateType: 'test_items', + aggregateLocalId: '019d-local-a', + operation: 'test_items.create', + payload: { title: 'Pending create A' }, + optimisticValue: { id: 0, title: 'Pending create A' }, + payloadHash: 'hash', + baseRevision: null, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + }); + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); + pull.mockResolvedValueOnce( + page([itemChange(99, 'Collision', { serverRevision: 2, acknowledgedCommandIds: ['cmd-create-a'] })], { nextCursor: 'cursor-v1' }), + ); + + await expect(service.pull(scope)).rejects.toThrow('Server id 99 is already mapped to another local replica row.'); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); + }); + }); + + it('acknowledgedCommandIds欠落changeは外部変更として受理する', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + localId: '019d-external', + serverId: 42, + values: { id: 42, title: 'Local baseline' }, + confirmedValues: { id: 42, title: 'Local baseline' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + }); + pull.mockResolvedValueOnce( + page( + [ + { + sourceKey: 'test_items', + serverId: 42, + serverRevision: 2, + values: { id: 42, title: 'Remote edit' }, + deleted: false, + }, + ], + { nextCursor: 'cursor-v1' }, + ), + ); + + await service.pull(scope); + + await expect(repository.getReplicaRow(scope, 'test_items', '019d-external')).resolves.toMatchObject({ + values: { title: 'Remote edit' }, + confirmedValues: { title: 'Remote edit' }, + serverRevision: 2, + syncState: 'confirmed', + }); + }); + + it('optional getCommandsForUser未実装repositoryでもpullがthrowしない', async () => { + TestBed.resetTestingModule(); + pull = vi.fn(async () => page([])); + TestBed.configureTestingModule({ + providers: [ + OfflineReplicaPullService, + { provide: OFFLINE_KIT_OPTIONS, useValue: { databaseName: 'test-offline', replicaSchema } }, + { provide: OFFLINE_REPLICA_PULLER, useValue: { pull } }, + { + provide: OFFLINE_COMMAND_HOOKS, + useValue: { entityType: (command: Pick) => command.aggregateType }, + }, + { + provide: OFFLINE_COMMAND_EXECUTOR, + useValue: { + execute: vi.fn(), + withServerRevision: (command: OfflineCommand, revision: string | number) => ({ + ...command, + baseRevision: revision, + }), + }, + }, + { + provide: OFFLINE_REPOSITORY, + useValue: { + getReplicaCursor: vi.fn(async () => null), + getCommands: vi.fn(async () => []), + transactReplica: vi.fn(async () => undefined), + getReplicaRow: vi.fn(async () => null), + getReplicaRowByServerId: vi.fn(async () => null), + }, + }, + ], + }); + service = TestBed.inject(OfflineReplicaPullService); + await expect(service.pull(scope)).resolves.toBeUndefined(); + }); + + it('non-finite numeric serverRevisionはrejectしcursorを進めない', async () => { + await expectPullRejectsPreservingCursor( + () => pull.mockResolvedValueOnce(page([{ ...itemChange(42, 'Created'), serverRevision: Number.NaN }], { nextCursor: 'cursor-v1' })), + 'Offline replica pull page changes[0].serverRevision must be a string or number.', + ); + }); +}); diff --git a/projects/kit/offline/src/lib/offline-replica-pull.service.ts b/projects/kit/offline/src/lib/offline-replica-pull.service.ts new file mode 100644 index 0000000..a317bf1 --- /dev/null +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -0,0 +1,304 @@ +import { inject, Injectable } from '@angular/core'; +import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; +import { OFFLINE_COMMAND_HOOKS } from './offline-command-hooks'; +import { OFFLINE_COMMAND_EXECUTOR } from './offline-command-executor'; +import { OFFLINE_REPLICA_PULLER, type OfflineReplicaChange, type OfflineReplicaPullPage } from './offline-replica-puller'; +import { projectOfflineReplicaValues, sha256OfflineReplicaSchema, type OfflineReplicaEntitySchema } from './offline-replica-schema'; +import { + OFFLINE_REPOSITORY, + type OfflineCommand, + type OfflineReplicaRow, + type OfflineReplicaRowKey, + type OfflineScope, +} from './offline-repository'; + +/** Pulls authoritative server deltas into one durable local replica partition. */ +@Injectable({ providedIn: 'root' }) +export class OfflineReplicaPullService { + readonly #repository = inject(OFFLINE_REPOSITORY); + readonly #options = inject(OFFLINE_KIT_OPTIONS); + readonly #puller = inject(OFFLINE_REPLICA_PULLER); + readonly #hooks = inject(OFFLINE_COMMAND_HOOKS); + readonly #executor = inject(OFFLINE_COMMAND_EXECUTOR); + #schemaHash: Promise | null = null; + + async pull(scope: OfflineScope): Promise { + const schemaHash = await (this.#schemaHash ??= sha256OfflineReplicaSchema(this.#options.replicaSchema)); + let cursor = (await this.#repository.getReplicaCursor(scope))?.cursor ?? ''; + + for (;;) { + const page = await this.#puller.pull({ + scope, + cursor, + schemaVersion: this.#options.replicaSchema.version, + schemaHash, + }); + this.#assertPullPage(page); + this.#assertHandshake(page.schemaVersion, page.schemaHash, schemaHash); + if (page.hasMore && page.nextCursor === cursor) { + throw new Error(`Offline replica pull cursor did not advance for scope ${scope.userId}:${scope.groupId}.`); + } + + const scopeCommands = await this.#repository.getCommands(scope); + const userCommands = this.#repository.getCommandsForUser ? await this.#repository.getCommandsForUser(scope.userId) : scopeCommands; + const changes = this.#collapseChanges(page.changes); + const putRows: OfflineReplicaRow[] = []; + const removeRows: OfflineReplicaRowKey[] = []; + const putCommands = new Map(); + const removeCommandIds = new Set(); + + for (const change of changes) { + const schema = this.#entitySchema(change.sourceKey); + const commands = schema.scope === 'user' ? userCommands : scopeCommands; + const acknowledged = (change.acknowledgedCommandIds ?? []) + .map((commandId) => { + const command = commands.find((candidate) => candidate.commandId === commandId); + if (!command) return null; + if (this.#hooks.entityType(command) !== change.sourceKey) { + throw new Error(`Acknowledged command "${commandId}" does not target "${change.sourceKey}".`); + } + return command; + }) + .filter((command): command is OfflineCommand => command !== null); + const acknowledgedLocalIds = new Set(acknowledged.map((command) => command.aggregateLocalId)); + if (acknowledgedLocalIds.size > 1) { + throw new Error(`Acknowledged commands for "${change.sourceKey}" target multiple local rows.`); + } + const acknowledgedCommand = acknowledged[0]; + const acknowledgedScope = acknowledgedCommand + ? { userId: acknowledgedCommand.userId, groupId: acknowledgedCommand.groupId } + : scope; + const acknowledgedRow = acknowledgedCommand + ? await this.#repository.getReplicaRow(acknowledgedScope, change.sourceKey, acknowledgedCommand.aggregateLocalId) + : null; + if (acknowledgedCommand && !acknowledgedRow) { + throw new Error(`Acknowledged command "${acknowledgedCommand.commandId}" has no local replica row.`); + } + const serverRow = await this.#repository.getReplicaRowByServerId(scope, change.sourceKey, change.serverId); + if (acknowledgedRow && serverRow && acknowledgedRow.localId !== serverRow.localId) { + throw new Error(`Server id ${change.serverId} is already mapped to another local replica row.`); + } + const existing = acknowledgedRow ?? serverRow; + const related = existing + ? commands.filter( + (command) => this.#hooks.entityType(command) === change.sourceKey && command.aggregateLocalId === existing.localId, + ) + : []; + const hasPending = related.length > 0; + + if (acknowledgedCommand) { + this.#applyAcknowledgement(change, existing!, related, putRows, removeRows, putCommands, removeCommandIds); + continue; + } + + if (change.deleted) { + if (!existing) continue; + if (!hasPending) { + removeRows.push(existing); + continue; + } + putRows.push({ ...existing, serverRevision: change.serverRevision, syncState: 'conflict', fetchedAt: Date.now() }); + for (const command of related) { + putCommands.set(command.commandId, { ...command, state: 'conflict', retryAt: null, lastErrorCode: 'remote_deleted' }); + } + continue; + } + + const confirmedValues = this.#validatedValues(schema, change); + if (!existing) { + putRows.push({ + ...scope, + sourceKey: change.sourceKey, + localId: crypto.randomUUID(), + serverId: change.serverId, + values: confirmedValues, + confirmedValues, + serverRevision: change.serverRevision, + fetchedAt: Date.now(), + syncState: 'confirmed', + }); + continue; + } + + const conflicted = related.some((command) => command.baseRevision !== change.serverRevision); + putRows.push({ + ...existing, + values: hasPending ? existing.values : confirmedValues, + confirmedValues, + serverRevision: change.serverRevision, + fetchedAt: Date.now(), + syncState: conflicted ? 'conflict' : hasPending ? 'pending' : 'confirmed', + }); + if (conflicted) { + for (const command of related) { + putCommands.set(command.commandId, { + ...command, + state: 'conflict', + retryAt: null, + lastErrorCode: 'remote_revision', + }); + } + } + } + + await this.#repository.transactReplica({ + putRows, + removeRows, + putCommands: [...putCommands.values()], + removeCommandIds: [...removeCommandIds], + putCursors: [{ ...scope, cursor: page.nextCursor }], + }); + cursor = page.nextCursor; + if (!page.hasMore) return; + } + } + + #assertPullPage(page: OfflineReplicaPullPage): void { + if (typeof page.nextCursor !== 'string') { + throw new Error('Offline replica pull page nextCursor must be a string.'); + } + if (typeof page.hasMore !== 'boolean') { + throw new Error('Offline replica pull page hasMore must be a boolean.'); + } + if (!Array.isArray(page.changes)) { + throw new Error('Offline replica pull page changes must be an array.'); + } + for (const [index, change] of page.changes.entries()) { + this.#assertPullChange(change, index); + } + } + + #assertPullChange(change: unknown, index: number): void { + const label = `Offline replica pull page changes[${index}]`; + if (!isPlainObject(change)) { + throw new Error(`${label} must be a plain object.`); + } + if (typeof change['sourceKey'] !== 'string') { + throw new Error(`${label}.sourceKey must be a string.`); + } + if (typeof change['deleted'] !== 'boolean') { + throw new Error(`${label}.deleted must be a boolean.`); + } + const serverId = change['serverId']; + if (typeof serverId !== 'number' || !Number.isFinite(serverId) || !Number.isSafeInteger(serverId) || serverId <= 0) { + throw new Error(`${label}.serverId must be a positive integer.`); + } + const revision = change['serverRevision']; + if (typeof revision !== 'string' && (typeof revision !== 'number' || !Number.isFinite(revision))) { + throw new Error(`${label}.serverRevision must be a string or number.`); + } + const acknowledgedCommandIds = change['acknowledgedCommandIds']; + if ( + (acknowledgedCommandIds !== undefined && !Array.isArray(acknowledgedCommandIds)) || + (Array.isArray(acknowledgedCommandIds) && + acknowledgedCommandIds.some((commandId) => typeof commandId !== 'string' || commandId.length === 0)) + ) { + throw new Error(`${label}.acknowledgedCommandIds must be an array of non-empty strings.`); + } + if (change['deleted']) { + if (change['values'] !== null) { + throw new Error(`${label} with deleted=true must have null values.`); + } + } + } + + #assertHandshake(version: number, hash: string, expectedHash: string): void { + if (version !== this.#options.replicaSchema.version || hash !== expectedHash) { + throw new Error( + `Offline replica schema mismatch: client=${this.#options.replicaSchema.version}/${expectedHash}, server=${version}/${hash}.`, + ); + } + } + + #entitySchema(sourceKey: string): OfflineReplicaEntitySchema> { + const schema = this.#options.replicaSchema.entities.find((entity) => entity.sourceKey === sourceKey); + if (!schema) throw new Error(`Unknown offline replica source key "${sourceKey}".`); + return schema; + } + + #validatedValues(schema: OfflineReplicaEntitySchema>, change: OfflineReplicaChange): unknown { + if (change.values === null) { + throw new Error(`Offline replica change "${change.sourceKey}"/${change.serverId} is missing values.`); + } + return projectOfflineReplicaValues(schema, change.values); + } + + #collapseChanges(changes: readonly OfflineReplicaChange[]): OfflineReplicaChange[] { + const collapsed = new Map(); + for (const change of changes) { + const key = `${change.sourceKey}:${change.serverId}`; + const previous = collapsed.get(key); + collapsed.set(key, { + ...change, + acknowledgedCommandIds: [...new Set([...(previous?.acknowledgedCommandIds ?? []), ...(change.acknowledgedCommandIds ?? [])])], + }); + } + return [...collapsed.values()]; + } + + #applyAcknowledgement( + change: OfflineReplicaChange, + row: OfflineReplicaRow, + related: readonly OfflineCommand[], + putRows: OfflineReplicaRow[], + removeRows: OfflineReplicaRowKey[], + putCommands: Map, + removeCommandIds: Set, + ): void { + const acknowledgedIds = new Set(change.acknowledgedCommandIds ?? []); + const lastAcknowledgedIndex = related.reduce((last, command, index) => (acknowledgedIds.has(command.commandId) ? index : last), -1); + if (lastAcknowledgedIndex < 0) { + throw new Error(`Replica acknowledgement does not match the local aggregate outbox.`); + } + if (related.slice(0, lastAcknowledgedIndex + 1).some((command) => !acknowledgedIds.has(command.commandId))) { + throw new Error(`Replica acknowledgement skipped an earlier aggregate command.`); + } + const following = related + .slice(lastAcknowledgedIndex + 1) + .map((command) => this.#executor.withServerRevision(command, change.serverRevision)); + for (const command of following) putCommands.set(command.commandId, command); + for (const command of related.slice(0, lastAcknowledgedIndex + 1)) { + removeCommandIds.add(command.commandId); + } + + if (change.deleted) { + if (following.length > 0) { + putRows.push({ ...row, serverRevision: change.serverRevision, syncState: 'conflict', fetchedAt: Date.now() }); + for (const command of following) { + putCommands.set(command.commandId, { ...command, state: 'conflict', lastErrorCode: 'remote_deleted' }); + } + } else { + removeRows.push(row); + } + return; + } + + const schema = this.#entitySchema(change.sourceKey); + const confirmedValues = this.#validatedValues(schema, change); + this.#assertServerIdAssignment(row.serverId, change.serverId); + putRows.push({ + ...row, + serverId: change.serverId, + values: following.length > 0 ? following.at(-1)!.optimisticValue : confirmedValues, + confirmedValues, + serverRevision: change.serverRevision, + fetchedAt: Date.now(), + syncState: following.length > 0 ? 'pending' : 'confirmed', + }); + } + + #assertServerIdAssignment(current: number | null, incoming: number): void { + if (current !== null && current !== incoming) { + throw new Error(`Replica serverId is immutable: current=${current}, incoming=${incoming}.`); + } + } +} + +function isPlainObject(value: unknown): value is Record { + if (value === null || typeof value !== 'object') { + return false; + } + + const prototype = Object.getPrototypeOf(value); + return prototype === Object.prototype || prototype === null; +} diff --git a/projects/kit/offline/src/lib/offline-replica-puller.ts b/projects/kit/offline/src/lib/offline-replica-puller.ts new file mode 100644 index 0000000..e3c3732 --- /dev/null +++ b/projects/kit/offline/src/lib/offline-replica-puller.ts @@ -0,0 +1,38 @@ +import { InjectionToken } from '@angular/core'; +import type { OfflineScope } from './offline-repository'; + +/** Server pull request for one user/group replica partition. */ +export interface OfflineReplicaPullRequest { + scope: OfflineScope; + cursor: string; + schemaVersion: number; + schemaHash: string; +} + +/** One server-side replica mutation returned by an explicit pull page. */ +export interface OfflineReplicaChange { + sourceKey: string; + serverId: number; + serverRevision: string | number; + /** Idempotency command ids durably recorded by the server and reflected in this final row state. */ + acknowledgedCommandIds?: readonly string[]; + values: unknown | null; + deleted: boolean; +} + +/** One explicit replica pull response page from the application backend. */ +export interface OfflineReplicaPullPage { + schemaVersion: number; + schemaHash: string; + changes: readonly OfflineReplicaChange[]; + nextCursor: string; + hasMore: boolean; +} + +/** Application-provided transport that fetches explicit replica pull pages from the server. */ +export interface OfflineReplicaPuller { + pull(request: OfflineReplicaPullRequest): Promise; +} + +/** DI token for the application-provided explicit replica pull transport. */ +export const OFFLINE_REPLICA_PULLER = new InjectionToken('OFFLINE_REPLICA_PULLER'); diff --git a/projects/kit/offline/src/lib/offline-replica-schema.spec.ts b/projects/kit/offline/src/lib/offline-replica-schema.spec.ts new file mode 100644 index 0000000..30748ab --- /dev/null +++ b/projects/kit/offline/src/lib/offline-replica-schema.spec.ts @@ -0,0 +1,744 @@ +/* eslint-disable @typescript-eslint/consistent-type-definitions */ +import { describe, expect, it } from 'vitest'; +import { + booleanColumn, + datetime, + decodeOfflineReplicaValues, + defineOfflineReplicaSchema, + defineReplicaEntity, + encodeOfflineReplicaValues, + ignored, + integer, + json, + nullable, + projectOfflineReplicaValues, + real, + serverId, + sha256OfflineReplicaSchema, + text, +} from './offline-replica-schema'; + +type SampleSelect = { + id: number; + title: string; + notes: string | null; + amount: number; + active: boolean; + payload: { version: number }; + updatedAt: string | Date; + transientFlag: boolean; +}; + +const sampleSchema = defineReplicaEntity()({ + table: 'sample_items', + sourceKey: 'sample_items', + scope: 'group', + fields: { + id: serverId(), + title: text(), + notes: nullable(text()), + amount: real(), + active: booleanColumn(), + payload: json<{ version: number }>(), + updatedAt: datetime(), + transientFlag: ignored('server-only cache flag'), + }, +}); + +describe('offline-replica-schema runtime', () => { + it('materializes ordered field descriptors with table metadata', () => { + expect(sampleSchema.tableName).toBe('sample_items'); + expect(sampleSchema.sourceKey).toBe('sample_items'); + expect(sampleSchema.scope).toBe('group'); + expect(sampleSchema.fields.map((field) => field.sourceKey)).toEqual([ + 'active', + 'amount', + 'id', + 'notes', + 'payload', + 'title', + 'transientFlag', + 'updatedAt', + ]); + expect(sampleSchema.fields.find((field) => field.sourceKey === 'id')).toEqual({ + sourceKey: 'id', + policy: 'serverId', + sqliteColumnName: 'server_id', + affinity: 'INTEGER', + storageKind: null, + nullable: true, + ignoredReason: null, + }); + expect(sampleSchema.fields.find((field) => field.sourceKey === 'notes')).toMatchObject({ + policy: 'column', + sqliteColumnName: 'notes', + affinity: 'TEXT', + storageKind: 'text', + nullable: true, + }); + expect(sampleSchema.fields.find((field) => field.sourceKey === 'updatedAt')).toMatchObject({ + policy: 'column', + sqliteColumnName: 'updated_at', + affinity: 'TEXT', + storageKind: 'datetime', + nullable: false, + }); + expect(sampleSchema.fields.find((field) => field.sourceKey === 'active')).toMatchObject({ + storageKind: 'booleanColumn', + affinity: 'INTEGER', + }); + expect(sampleSchema.fields.find((field) => field.sourceKey === 'amount')).toMatchObject({ + storageKind: 'real', + affinity: 'REAL', + }); + expect(sampleSchema.fields.find((field) => field.sourceKey === 'payload')).toMatchObject({ + storageKind: 'json', + affinity: 'TEXT', + }); + expect(sampleSchema.fields.find((field) => field.sourceKey === 'transientFlag')).toMatchObject({ + policy: 'ignored', + sqliteColumnName: null, + ignoredReason: 'server-only cache flag', + }); + }); + + it('generates deterministic CREATE TABLE SQL and scoped server_id index', () => { + expect(sampleSchema.createTableSql).toEqual([ + `CREATE TABLE IF NOT EXISTS sample_items ( + local_id TEXT NOT NULL, + _offline_user_id INTEGER NOT NULL, + _offline_group_id INTEGER NOT NULL, + server_id INTEGER, + _offline_confirmed_json TEXT, + _offline_server_revision_json TEXT, + _offline_sync_state TEXT NOT NULL, + _offline_fetched_at INTEGER NOT NULL, + active INTEGER NOT NULL, + amount REAL NOT NULL, + notes TEXT, + payload TEXT NOT NULL, + title TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (local_id) +)`, + 'CREATE UNIQUE INDEX IF NOT EXISTS uq_sample_items_server_id ON sample_items (_offline_user_id, _offline_group_id, server_id) WHERE server_id IS NOT NULL', + ]); + }); + + it('omits _offline_group_id and scopes the partial index to _offline_user_id for user scope', () => { + type UserScopedSelect = { id: number; title: string }; + const userScopedSchema = defineReplicaEntity()({ + table: 'user_notes', + sourceKey: 'user_notes', + scope: 'user', + fields: { + id: serverId(), + title: text(), + }, + }); + + expect(userScopedSchema.createTableSql).toEqual([ + `CREATE TABLE IF NOT EXISTS user_notes ( + local_id TEXT NOT NULL, + _offline_user_id INTEGER NOT NULL, + server_id INTEGER, + _offline_confirmed_json TEXT, + _offline_server_revision_json TEXT, + _offline_sync_state TEXT NOT NULL, + _offline_fetched_at INTEGER NOT NULL, + title TEXT NOT NULL, + PRIMARY KEY (local_id) +)`, + 'CREATE UNIQUE INDEX IF NOT EXISTS uq_user_notes_server_id ON user_notes (_offline_user_id, server_id) WHERE server_id IS NOT NULL', + ]); + }); + + it('exposes a deterministic schema fingerprint input', () => { + expect(sampleSchema.schemaFingerprintInput).toBe( + 'table=sample_items|source=sample_items|scope=group|hasServerId=1|fields=active:column:active:INTEGER:booleanColumn:required;amount:column:amount:REAL:real:required;id:serverId:server_id:INTEGER:nullable;notes:column:notes:TEXT:text:nullable;payload:column:payload:TEXT:json:required;title:column:title:TEXT:text:required;transientFlag:ignored:server-only cache flag;updatedAt:column:updated_at:TEXT:datetime:required', + ); + }); + + it('rejects invalid table and reserved column identifiers', () => { + type MinimalSelect = { id: number; title: string }; + expect(() => + defineReplicaEntity()({ + table: 'Bad-Table', + sourceKey: 'items', + scope: 'user', + fields: { id: serverId(), title: text() }, + }), + ).toThrow('Replica table "Bad-Table" must match ^[a-z][a-z0-9_]*$.'); + + expect(() => + defineReplicaEntity()({ + table: 'items', + sourceKey: 'items', + scope: 'user', + fields: { + id: serverId(), + title: { kind: 'column', affinity: 'TEXT', storageKind: 'text', columnName: 'local_id', nullable: false }, + }, + }), + ).toThrow('Replica column "local_id" is reserved.'); + }); + + it('rejects ignored fields with an empty reason', () => { + type Select = { id: number; flag: boolean }; + expect(() => + defineReplicaEntity()({ + table: 'items', + sourceKey: 'items', + scope: 'user', + fields: { id: serverId(), altId: serverId() }, + }), + ).toThrow('Replica entity must define exactly one serverId field.'); + }); + + it('rejects zero serverId fields', () => { + type Select = { title: string }; + expect(() => + defineReplicaEntity