1import {
2 UiSettingsProfile,
3 UiSettingsValues,
4 UiSettingsPatchInput,
5 type UiSettingsPatchResult,
6 EnvironmentId,
7 WS_METHODS,
8} from "@t3tools/contracts";
9import * as Equal from "effect/Equal";
10import type * as EnvironmentRegistry from "../connection/registry.ts";
11import type { Atom } from "effect/reactivity";
12import * as Schema from "effect/Schema";
13import {
14 createEnvironmentRpcCommand,
15 createEnvironmentRpcSubscriptionAtomFamily,
16} from "./runtime.ts";
17
18export const UiSettingsSyncCache = Schema.Struct({
19 environmentId: EnvironmentId,
20 profile: UiSettingsProfile,
21 pending: UiSettingsPatchInput,
22});
23export type UiSettingsSyncCache = typeof UiSettingsSyncCache.Type;
24
25/** Platform adapters keep their existing preference stores and apply only supported fields. */
26export class UiSettingsSync {
27 private cache: UiSettingsSyncCache;
28 private connectionGeneration = 0;
29 private stopped = false;
30 private flushing = false;
31 private persistence: Promise<void> = Promise.resolve();
32 private listeners = new Set<() => void>();
33 private status: "offline" | "syncing" | "synced" | "error" = "offline";
34 private conflicts: ReadonlyArray<string> = [];
35
36 private readonly adapter: {
37 readonly save: (cache: UiSettingsSyncCache) => Promise<void>;
38 readonly apply: (values: UiSettingsValues) => Promise<void>;
39 readonly patch: (input: UiSettingsPatchInput) => Promise<UiSettingsPatchResult>;
40 };
41
42 constructor(cache: UiSettingsSyncCache, adapter: UiSettingsSync["adapter"]) {
43 this.cache = cache;
44 this.adapter = adapter;
45 }
46
47 get snapshot() {
48 return { status: this.status, conflicts: this.conflicts };
49 }
50
51 readonly subscribe = (listener: () => void) => {
52 this.listeners.add(listener);
53 return () => {
54 this.listeners.delete(listener);
55 };
56 };
57
58 private notify() {
59 for (const listener of this.listeners) listener();
60 }
61
62 private save() {
63 const cache = this.cache;
64 const saved = this.persistence.then(() => this.adapter.save(cache));
65 this.persistence = saved.catch(() => undefined);
66 return saved;
67 }
68
69 private get online() {
70 return this.status === "syncing" || this.status === "synced";
71 }
72
73 private get values(): UiSettingsValues {
74 return { ...this.cache.profile.values, ...this.cache.pending.patch };
75 }
76
77 /** Records deliberate edits only; hydration and remote applications never call this. */
78 edit(patch: UiSettingsValues) {
79 if (this.stopped) return;
80 const pending = { ...this.cache.pending.patch };
81 const baseRevisions = { ...this.cache.pending.baseRevisions };
82 let changed = false;
83 for (const key of Object.keys(patch) as Array<keyof UiSettingsValues>) {
84 if (Equal.equals(this.values[key], patch[key])) continue;
85 Object.assign(pending, { [key]: patch[key] });
86 baseRevisions[key] ??= this.cache.profile.fieldRevisions[key] ?? 0;
87 changed = true;
88 }
89 if (!changed) return;
90 this.cache = { ...this.cache, pending: { patch: pending, baseRevisions } };
91 this.conflicts = [];
92 this.status = this.online ? "syncing" : "offline";
93 this.notify();
94 void this.save()
95 .then(() => this.flush())
96 .catch(() => this.fail());
97 }
98
99 async restore() {
100 if (this.stopped) return;
101 try {
102 await this.adapter.apply(this.values);
103 } catch {
104 this.fail();
105 }
106 }
107
108 async receive(profile: UiSettingsProfile) {
109 if (this.stopped) return;
110 if (profile.revision < this.cache.profile.revision) return;
111 const generation = this.connectionGeneration;
112 this.cache = { ...this.cache, profile };
113 try {
114 await this.save();
115 if (this.stopped) return;
116 await this.adapter.apply(this.values);
117 if (this.stopped || generation !== this.connectionGeneration) return;
118 this.status = "syncing";
119 await this.flush();
120 } catch {
121 if (generation === this.connectionGeneration) this.fail();
122 }
123 }
124
125 disconnect() {
126 this.connectionGeneration += 1;
127 this.status = "offline";
128 this.notify();
129 }
130
131 stop() {
132 this.stopped = true;
133 this.disconnect();
134 return this.persistence;
135 }
136
137 private fail() {
138 if (this.stopped) return;
139 this.status = "error";
140 this.notify();
141 }
142
143 private async flush() {
144 if (this.flushing || !this.online || this.stopped) return;
145 const generation = this.connectionGeneration;
146 this.flushing = true;
147 try {
148 while (this.online && !this.stopped) {
149 const sent = this.cache.pending;
150 if (Object.keys(sent.patch).length === 0) break;
151 this.status = "syncing";
152 this.notify();
153 const result = await this.adapter.patch(sent);
154 if (this.stopped) return;
155 const patch = { ...this.cache.pending.patch };
156 const baseRevisions = { ...this.cache.pending.baseRevisions };
157 for (const key of Object.keys(sent.patch) as Array<keyof UiSettingsValues>) {
158 if (Equal.equals(patch[key], sent.patch[key])) {
159 delete patch[key];
160 delete baseRevisions[key];
161 } else if (
162 !result.conflicts.includes(key) &&
163 baseRevisions[key] === sent.baseRevisions[key]
164 ) {
165 baseRevisions[key] = result.profile.fieldRevisions[key] ?? 0;
166 }
167 }
168 this.conflicts = result.conflicts;
169 const profile =
170 result.profile.revision >= this.cache.profile.revision
171 ? result.profile
172 : this.cache.profile;
173 this.cache = { ...this.cache, profile, pending: { patch, baseRevisions } };
174 await this.save();
175 if (this.stopped) return;
176 await this.adapter.apply(this.values);
177 }
178 if (this.online && !this.stopped) this.status = "synced";
179 } catch {
180 if (generation === this.connectionGeneration) this.fail();
181 } finally {
182 this.flushing = false;
183 this.notify();
184 if (generation !== this.connectionGeneration && this.online && !this.stopped) {
185 await this.flush();
186 }
187 }
188 }
189}
190
191export function createUiSettingsSyncAtoms<R, E>(
192 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, E>,
193) {
194 return {
195 read: createEnvironmentRpcCommand(runtime, {
196 label: "ui-settings:read",
197 tag: WS_METHODS.uiSettingsGet,
198 }),
199 patch: createEnvironmentRpcCommand(runtime, {
200 label: "ui-settings:patch",
201 tag: WS_METHODS.uiSettingsPatch,
202 }),
203 changes: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
204 label: "ui-settings:changes",
205 tag: WS_METHODS.uiSettingsSubscribe,
206 idleTtlMs: 0,
207 }),
208 };
209}