import { withHostsSanitizedForSync, type CloudSyncPayloadEntityKey, type SyncFileMeta, type SyncPayload, type SyncReliabilityMeta, } from '../sync'; import { dotKey } from './clock'; import { cloneJson, isJsonValue, jsonValuesEqual, normalizeJsonValue, } from './json'; import { selectRegisterWinner, isTombstoneCandidate } from './register'; import { createEmptyRecord, setOwnRecordValue } from './record'; import { assertValidConvergentSyncState, canonicalizeConvergentSyncState, decodeSettingPath, } from './serialization'; import { applyConvergentMutations, createConvergentSyncState, materializeConvergentSyncState, } from './state'; import type { CollectionPosition, ConvergentEnvelopeCandidate, ConvergentEnvelopeCollectionState, ConvergentEnvelopeEntityState, ConvergentEnvelopeRegister, ConvergentEnvelopeStateV2, ConvergentEnvelopeStringCollectionState, ConvergentEnvelopeStringEntryState, ConvergentMutation, ConvergentSyncEnvelopeV2, ConvergentSyncStateV2, JsonObject, JsonValue, MultiValueRegister, RegisterCandidate, } from './types'; export const CONVERGENT_ENTITY_COLLECTIONS = [ 'hosts', 'keys', 'identities', 'proxyProfiles', 'snippets', 'notes', 'portForwardingRules', 'groupConfigs', ] as const satisfies readonly CloudSyncPayloadEntityKey[]; export const CONVERGENT_STRING_COLLECTIONS = [ 'customGroups', 'snippetPackages', 'noteGroups', ] as const satisfies readonly CloudSyncPayloadEntityKey[]; type ConvergentEntityCollection = typeof CONVERGENT_ENTITY_COLLECTIONS[number]; type ConvergentStringCollection = typeof CONVERGENT_STRING_COLLECTIONS[number]; const ENTITY_COLLECTION_SET = new Set(CONVERGENT_ENTITY_COLLECTIONS); const STRING_COLLECTION_SET = new Set(CONVERGENT_STRING_COLLECTIONS); function isRecord(value: unknown): value is Record { return Boolean(value) && typeof value === 'object' && !Array.isArray(value); } function toJsonValue(value: unknown, label: string): JsonValue { try { return normalizeJsonValue(value); } catch { throw new Error(`${label} contains a value that cannot be represented as JSON`); } } function entityId(collection: ConvergentEntityCollection, value: Record): string { const raw = collection === 'groupConfigs' ? value.path : value.id; if (typeof raw !== 'string' || raw.length === 0) { throw new Error(`${collection} contains an entity without a stable identifier`); } return raw; } function entityJson( collection: ConvergentEntityCollection, value: Record, ): JsonObject { const id = entityId(collection, value); const json = toJsonValue(value, `${collection}/${id}`); if (!isRecord(json)) throw new Error(`${collection}/${id} must be a JSON object`); return { ...json, id, } as JsonObject; } function payloadEntityValues( payload: SyncPayload, collection: ConvergentEntityCollection, ): Record[] { const values = payload[collection]; return Array.isArray(values) ? values as unknown as Record[] : []; } function payloadStringValues( payload: SyncPayload, collection: ConvergentStringCollection, ): string[] { const values = payload[collection]; return Array.isArray(values) ? values.filter((value): value is string => typeof value === 'string') : []; } function appendSettingMutations( value: unknown, path: string[], mutations: ConvergentMutation[], ): void { if (isRecord(value) && Object.keys(value).length > 0) { for (const key of Object.keys(value).sort()) { appendSettingMutations(value[key], [...path, key], mutations); } return; } if (path.length === 0 || value === undefined) return; mutations.push({ kind: 'setting-set', path, value: toJsonValue(value, `settings.${path.join('.')}`), }); } export function syncPayloadToConvergentMutations(payload: SyncPayload): ConvergentMutation[] { const sanitized = withHostsSanitizedForSync(payload); const mutations: ConvergentMutation[] = []; for (const collection of CONVERGENT_ENTITY_COLLECTIONS) { payloadEntityValues(sanitized, collection).forEach((value, position) => { const id = entityId(collection, value); mutations.push({ kind: 'entity-upsert', collection, entityId: id, value: entityJson(collection, value), position, }); }); } for (const collection of CONVERGENT_STRING_COLLECTIONS) { payloadStringValues(sanitized, collection).forEach((value, position) => { mutations.push({ kind: 'string-entry-add', collection, value, position }); }); } appendSettingMutations(sanitized.settings, [], mutations); return mutations; } export function createConvergentSyncStateFromPayload( payload: SyncPayload, deviceId: string, now: number, ): ConvergentSyncStateV2 { return applyConvergentMutations( createConvergentSyncState(), deviceId, syncPayloadToConvergentMutations(payload), now, ); } function requireKnownCollections(state: ConvergentSyncStateV2): void { for (const collection of Object.keys(state.collections)) { if (!ENTITY_COLLECTION_SET.has(collection)) { throw new Error(`Unsupported convergent entity collection: ${collection}`); } } for (const collection of Object.keys(state.stringCollections)) { if (!STRING_COLLECTION_SET.has(collection)) { throw new Error(`Unsupported convergent string collection: ${collection}`); } } } function collectionValues( collections: Record, collection: ConvergentEntityCollection, ): JsonObject[] { return collections[collection] ?? []; } function typedCollection( collections: Record, collection: Exclude, ): T[] { return collectionValues(collections, collection) as unknown as T[]; } export function materializeSyncPayloadFromConvergentState( state: ConvergentSyncStateV2, options: { syncedAt: number; syncMeta?: SyncReliabilityMeta; /** Opaque plugin sidecars travel with the encrypted blob outside CRDT fields. */ pluginSidecars?: SyncPayload['pluginSidecars']; }, ): SyncPayload { requireKnownCollections(state); const materialized = materializeConvergentSyncState(state); const groupConfigs = collectionValues(materialized.collections, 'groupConfigs').map((value) => { const { id: _id, ...groupConfig } = value; return groupConfig as unknown as import('../models').GroupConfig; }); const settings = Object.keys(materialized.settings).length > 0 ? materialized.settings as unknown as NonNullable : undefined; return { hosts: typedCollection(materialized.collections, 'hosts'), keys: typedCollection(materialized.collections, 'keys'), identities: typedCollection(materialized.collections, 'identities'), proxyProfiles: typedCollection(materialized.collections, 'proxyProfiles'), snippets: typedCollection(materialized.collections, 'snippets'), customGroups: materialized.stringCollections.customGroups ?? [], snippetPackages: materialized.stringCollections.snippetPackages ?? [], notes: typedCollection(materialized.collections, 'notes'), noteGroups: materialized.stringCollections.noteGroups ?? [], portForwardingRules: typedCollection(materialized.collections, 'portForwardingRules'), groupConfigs, settings, syncedAt: options.syncedAt, ...(options.syncMeta ? { syncMeta: options.syncMeta } : {}), // Preserve explicit empty bundles so lifecycle materializations (conflict // resolve / downgrade) can clear or re-upload sidecars rather than omit // the field and look like a legacy payload. ...(options.pluginSidecars && Array.isArray(options.pluginSidecars.entries) ? { pluginSidecars: { version: 1 as const, entries: options.pluginSidecars.entries, }, } : {}), }; } function materializedEntity( payload: SyncPayload, collection: string, id: string, ): Record | undefined { if (!ENTITY_COLLECTION_SET.has(collection)) return undefined; const values = payloadEntityValues(payload, collection as ConvergentEntityCollection); return values.find((value) => entityId(collection as ConvergentEntityCollection, value) === id); } function nestedSetting(payload: SyncPayload, path: string[]): unknown { let value: unknown = payload.settings; for (const segment of path) { if (!isRecord(value) || !Object.prototype.hasOwnProperty.call(value, segment)) return undefined; value = value[segment]; } return value; } function stableUnknown(value: unknown): unknown { if (Array.isArray(value)) return value.map(stableUnknown); if (isRecord(value)) { return Object.fromEntries( Object.keys(value).sort().map((key) => [key, stableUnknown(value[key])]), ); } return value; } function materializedCloudFingerprint(payload: SyncPayload): string { return JSON.stringify(stableUnknown({ ...Object.fromEntries( CONVERGENT_ENTITY_COLLECTIONS.map((collection) => [ collection, payloadEntityValues(payload, collection), ]), ), ...Object.fromEntries( CONVERGENT_STRING_COLLECTIONS.map((collection) => [ collection, payloadStringValues(payload, collection), ]), ), settings: payload.settings ?? {}, })); } function assertMaterializedPayloadMatchesState( state: ConvergentSyncStateV2, payload: SyncPayload, ): void { const expected = materializeSyncPayloadFromConvergentState(state, { syncedAt: 0 }); if (materializedCloudFingerprint(expected) !== materializedCloudFingerprint(payload)) { throw new Error('Convergent envelope does not match its materialized v1 snapshot'); } } function compactRegister( register: MultiValueRegister, materializedValue?: unknown, allowMaterializedValue = false, ): ConvergentEnvelopeRegister { const winner = selectRegisterWinner(register); return { candidates: register.candidates.map((candidate): ConvergentEnvelopeCandidate => { const base = { dot: { ...candidate.dot }, context: candidate.context.map((dot) => ({ ...dot })), hlc: { ...candidate.hlc }, }; if (isTombstoneCandidate(candidate)) return { ...base, tombstone: true }; if ( allowMaterializedValue && winner && dotKey(candidate.dot) === dotKey(winner.dot) && isJsonValue(materializedValue) && jsonValuesEqual(candidate.value, materializedValue) ) { return { ...base, materialized: true }; } return { ...base, value: cloneJson(candidate.value) }; }), }; } export function createConvergentSyncEnvelope( state: ConvergentSyncStateV2, materializedPayload: SyncPayload, ): ConvergentSyncEnvelopeV2 { const canonical = canonicalizeConvergentSyncState(state); requireKnownCollections(canonical); assertMaterializedPayloadMatchesState(canonical, materializedPayload); const collections = createEmptyRecord(); for (const [collectionName, collection] of Object.entries(canonical.collections)) { const entities = createEmptyRecord(); for (const [id, entity] of Object.entries(collection.entities)) { const materialized = materializedEntity(materializedPayload, collectionName, id); const fields = createEmptyRecord(); for (const [field, register] of Object.entries(entity.fields)) { setOwnRecordValue(fields, field, compactRegister(register, materialized?.[field], true)); } setOwnRecordValue(entities, id, { presence: compactRegister(entity.presence), ...(entity.position ? { position: compactRegister(entity.position) } : {}), fields, }); } setOwnRecordValue(collections, collectionName, { entities }); } const settings = createEmptyRecord(); for (const [encodedPath, register] of Object.entries(canonical.settings)) { setOwnRecordValue(settings, encodedPath, compactRegister( register, nestedSetting(materializedPayload, decodeSettingPath(encodedPath)), true, )); } const stringCollections = createEmptyRecord(); for (const [collectionName, collection] of Object.entries(canonical.stringCollections)) { const entries = createEmptyRecord(); for (const [value, entry] of Object.entries(collection.entries)) { setOwnRecordValue(entries, value, { presence: compactRegister(entry.presence), ...(entry.position ? { position: compactRegister(entry.position) } : {}), }); } setOwnRecordValue(stringCollections, collectionName, { entries }); } return { schemaVersion: 2, encoding: 'materialized-winner-v1', state: { vector: Object.fromEntries(Object.entries(canonical.vector)), dotOrigins: Object.fromEntries( Object.entries(canonical.dotOrigins).map(([deviceId, origins]) => [deviceId, { ...origins }]), ), hlc: { ...canonical.hlc }, collections, settings, stringCollections, }, }; } function hydrateRegister( register: ConvergentEnvelopeRegister, materializedValue: unknown, label: string, ): MultiValueRegister { if (!register || !Array.isArray(register.candidates) || register.candidates.length === 0) { throw new Error(`${label} has no candidates`); } return { candidates: register.candidates.map((candidate, index): RegisterCandidate => { const candidateLabel = `${label}.candidates[${index}]`; const base = { dot: { ...candidate.dot }, context: candidate.context.map((dot) => ({ ...dot })), hlc: { ...candidate.hlc }, }; if ( candidate.tombstone !== undefined && candidate.tombstone !== true && candidate.tombstone !== false ) { throw new Error(`${candidateLabel} has an invalid tombstone marker`); } if ( 'materialized' in candidate && candidate.materialized !== undefined && candidate.materialized !== true ) { throw new Error(`${candidateLabel} has an invalid materialized marker`); } if (candidate.tombstone === true) { if ('materialized' in candidate || 'value' in candidate) { throw new Error(`${candidateLabel} tombstone contains a value marker`); } return { ...base, tombstone: true }; } if ('materialized' in candidate && candidate.materialized === true) { if ('value' in candidate || !isJsonValue(materializedValue)) { throw new Error(`${candidateLabel} cannot reconstruct its materialized value`); } return { ...base, value: cloneJson(materializedValue) as T }; } if (!('value' in candidate) || !isJsonValue(candidate.value)) { throw new Error(`${candidateLabel} is missing a JSON value`); } return { ...base, value: cloneJson(candidate.value) as T }; }), }; } function envelopeState(value: unknown): ConvergentEnvelopeStateV2 { if (!isRecord(value)) throw new Error('Convergent sync envelope state is invalid'); return value as unknown as ConvergentEnvelopeStateV2; } export function hydrateConvergentSyncEnvelope( envelope: ConvergentSyncEnvelopeV2, materializedPayload: SyncPayload, ): ConvergentSyncStateV2 { if ( !envelope || envelope.schemaVersion !== 2 || envelope.encoding !== 'materialized-winner-v1' ) { throw new Error('Unsupported convergent sync envelope'); } const encoded = envelopeState(envelope.state); const collections = createEmptyRecord(); for (const [collectionName, collection] of Object.entries(encoded.collections ?? {})) { if (!ENTITY_COLLECTION_SET.has(collectionName) || !isRecord(collection?.entities)) { throw new Error(`Unsupported or invalid convergent collection: ${collectionName}`); } const entities = createEmptyRecord(); for (const [id, entity] of Object.entries(collection.entities)) { if (!isRecord(entity) || !isRecord(entity.fields)) { throw new Error(`Invalid convergent entity: ${collectionName}/${id}`); } const materialized = materializedEntity(materializedPayload, collectionName, id); const fields = createEmptyRecord(); for (const [field, register] of Object.entries(entity.fields)) { setOwnRecordValue(fields, field, hydrateRegister( register, materialized?.[field], `${collectionName}/${id}/${field}`, )); } setOwnRecordValue(entities, id, { presence: hydrateRegister(entity.presence, true, `${collectionName}/${id}/presence`), ...(entity.position ? { position: hydrateRegister(entity.position, undefined, `${collectionName}/${id}/position`) } : {}), fields, }); } setOwnRecordValue(collections, collectionName, { entities }); } const settings = createEmptyRecord(); for (const [path, register] of Object.entries(encoded.settings ?? {})) { setOwnRecordValue(settings, path, hydrateRegister( register, nestedSetting(materializedPayload, decodeSettingPath(path)), `settings/${path}`, )); } const stringCollections = createEmptyRecord(); for (const [collectionName, collection] of Object.entries(encoded.stringCollections ?? {})) { if (!STRING_COLLECTION_SET.has(collectionName) || !isRecord(collection?.entries)) { throw new Error(`Unsupported or invalid convergent string collection: ${collectionName}`); } const entries = createEmptyRecord(); for (const [value, entry] of Object.entries(collection.entries)) { if (!isRecord(entry)) throw new Error(`Invalid convergent string entry: ${collectionName}/${value}`); setOwnRecordValue(entries, value, { presence: hydrateRegister(entry.presence, true, `${collectionName}/${value}/presence`), ...(entry.position ? { position: hydrateRegister(entry.position, undefined, `${collectionName}/${value}/position`) } : {}), }); } setOwnRecordValue(stringCollections, collectionName, { entries }); } const state: ConvergentSyncStateV2 = { schemaVersion: 2, vector: encoded.vector, dotOrigins: encoded.dotOrigins, hlc: encoded.hlc, collections, settings, stringCollections, }; assertValidConvergentSyncState(state); const canonical = canonicalizeConvergentSyncState(state); assertMaterializedPayloadMatchesState(canonical, materializedPayload); return canonical; } export function withConvergentSyncEnvelope( state: ConvergentSyncStateV2, options: { syncedAt: number; syncMeta?: SyncReliabilityMeta; pluginSidecars?: SyncPayload['pluginSidecars']; }, ): SyncPayload { const payload = materializeSyncPayloadFromConvergentState(state, options); return { ...payload, convergentSync: createConvergentSyncEnvelope(state, payload), }; } export function validateConvergentSyncPayload( meta: Pick, payload: SyncPayload, ): ConvergentSyncStateV2 | null { const schemaVersion = (meta as { syncSchemaVersion?: unknown }).syncSchemaVersion; if (schemaVersion === undefined) { if (payload.convergentSync !== undefined) { throw new Error('Convergent sync envelope is present without schema metadata'); } return null; } if (schemaVersion !== 2) { throw new Error(`Unsupported sync schema version: ${String(schemaVersion)}`); } if (!payload.convergentSync) { throw new Error('Sync schema v2 payload is missing its convergent envelope'); } return hydrateConvergentSyncEnvelope(payload.convergentSync, payload); } /** Prevent the legacy snapshot writer from silently erasing v2/future metadata. */ export function assertConvergentSyncWriteCompatible( remoteMeta: Pick | null | undefined, outgoingPayload: SyncPayload, ): void { if (!remoteMeta) return; const remoteSchema = (remoteMeta as { syncSchemaVersion?: unknown }).syncSchemaVersion; if (remoteSchema === undefined) return; if (remoteSchema !== 2) { throw new Error(`Cannot overwrite unsupported sync schema version: ${String(remoteSchema)}`); } if (!outgoingPayload.convergentSync) { throw new Error( 'Cloud data uses convergent sync v2. Enable or migrate convergent sync before uploading.', ); } } export function stripConvergentSyncEnvelope(payload: SyncPayload): SyncPayload { const { convergentSync: _convergentSync, ...legacyPayload } = payload; return legacyPayload; }