forked from
tokono.ma/diffuse
A music player that connects to your cloud/distributed storage.
wip: dasl sync
+335
+314
src/components/transformer/output/bytes/dasl-sync/element.js
+314
src/components/transformer/output/bytes/dasl-sync/element.js
···
1
+
import { Temporal } from "@js-temporal/polyfill";
2
+
import { ifDefined } from "lit-html/directives/if-defined.js";
3
+
import { decode, encode } from "@atcute/cbor";
4
+
import deepDiff from "@fry69/deep-diff";
5
+
6
+
import "@components/output/polymorphic/indexed-db/element.js";
7
+
8
+
import * as CID from "@common/cid.js";
9
+
import { computed, signal } from "@common/signal.js";
10
+
import { OutputTransformer } from "../../base.js";
11
+
12
+
/**
13
+
* @import { RenderArg } from "@common/element.d.ts"
14
+
* @import { SignalReader } from "@common/signal.d.ts";
15
+
* @import { OutputElement } from "@components/output/types.d.ts";
16
+
*
17
+
* @import { Container } from "./types.d.ts"
18
+
*/
19
+
20
+
/**
21
+
* @extends {OutputTransformer<Uint8Array>}
22
+
*/
23
+
class DaslBytesSyncOutputTransformer extends OutputTransformer {
24
+
constructor() {
25
+
super();
26
+
27
+
const remote = this.base();
28
+
const local = this.#localOutput.get;
29
+
30
+
/**
31
+
* @template {{ id: string }} T
32
+
* @param {SignalReader<Uint8Array | undefined>} localCollection
33
+
* @param {SignalReader<Uint8Array | undefined>} remoteCollection
34
+
* @returns {SignalReader<{ container: Container<T>; diverged: boolean; local: boolean; remote: boolean; }>}
35
+
*/
36
+
const state = (localCollection, remoteCollection) =>
37
+
computed(() => {
38
+
const lb = localCollection();
39
+
const rb = remote.ready() ? remoteCollection() : undefined;
40
+
41
+
/** @type {Container<T> | undefined} */
42
+
const l = lb ? decode(lb) : undefined;
43
+
44
+
/** @type {Container<T> | undefined} */
45
+
const r = rb ? decode(rb) : undefined;
46
+
47
+
if (!r) {
48
+
return l
49
+
? {
50
+
container: l,
51
+
diverged: remote.ready(),
52
+
local: false,
53
+
remote: remote.ready(),
54
+
}
55
+
: {
56
+
container: {
57
+
cid: "",
58
+
data: [],
59
+
inventory: { current: {}, removed: [] },
60
+
},
61
+
diverged: false,
62
+
local: false,
63
+
remote: false,
64
+
};
65
+
} else if (!l) {
66
+
return { container: r, diverged: true, local: true, remote: false };
67
+
}
68
+
69
+
const diverged = this.hasDiverged({ local: l, remote: r });
70
+
71
+
return {
72
+
container: diverged.local || diverged.remote
73
+
? /* this.merge(l, r) */ l
74
+
: r,
75
+
diverged: diverged.local || diverged.remote,
76
+
local: diverged.local,
77
+
remote: diverged.remote,
78
+
};
79
+
});
80
+
81
+
const facets = state(
82
+
computed(() => local()?.facets?.collection()),
83
+
remote.facets.collection,
84
+
);
85
+
86
+
const playlistItems = state(
87
+
computed(() => local()?.playlistItems?.collection()),
88
+
remote.playlistItems.collection,
89
+
);
90
+
91
+
const themes = state(
92
+
computed(() => local()?.themes?.collection()),
93
+
remote.themes.collection,
94
+
);
95
+
96
+
const tracks = state(
97
+
computed(() => local()?.tracks?.collection()),
98
+
remote.tracks.collection,
99
+
);
100
+
101
+
this.facets = undefined;
102
+
this.playlistItems = undefined;
103
+
this.themes = undefined;
104
+
this.tracks = undefined;
105
+
106
+
this.ready = () => true;
107
+
108
+
// Effects
109
+
}
110
+
111
+
// SIGNALS
112
+
113
+
#localOutput = signal(
114
+
/** @type {OutputElement<Uint8Array | undefined> | undefined} */ (undefined),
115
+
);
116
+
117
+
// LIFECYCLE
118
+
119
+
/**
120
+
* @override
121
+
*/
122
+
connectedCallback() {
123
+
super.connectedCallback();
124
+
125
+
/** @type {OutputElement<Uint8Array | undefined> | null} */
126
+
const local = this.root().querySelector("dop-indexed-db");
127
+
if (!local) throw new Error("Can't find local output");
128
+
129
+
// When defined
130
+
customElements.whenDefined(local.localName).then(() => {
131
+
this.#localOutput.value = local;
132
+
});
133
+
}
134
+
135
+
// 🛠️
136
+
137
+
/**
138
+
* @template {{ id: string }} T
139
+
* @param {{ local: Container<T>, remote: Container<T> }} _
140
+
* @returns {{ local: boolean, remote: boolean }} Which store needs updating?
141
+
*/
142
+
hasDiverged({ local, remote }) {
143
+
const diverged = local.cid !== remote.cid;
144
+
145
+
if (!diverged) {
146
+
return {
147
+
local: false,
148
+
remote: false,
149
+
};
150
+
}
151
+
152
+
// TODO: Could be improved.
153
+
// We might not need to save on both ends.
154
+
return {
155
+
local: true,
156
+
remote: true,
157
+
};
158
+
}
159
+
160
+
/**
161
+
* @template {{ id: string; updatedAt: string }} T
162
+
* @param {Container<T>} a
163
+
* @param {Container<T>} b
164
+
* @returns {Promise<Container<T>>}
165
+
*/
166
+
async merge(a, b) {
167
+
const removedA = new Set(a.inventory.removed);
168
+
const removedB = new Set(b.inventory.removed);
169
+
const allRemoved = removedA.union(removedB);
170
+
171
+
const currentA = a.inventory.current;
172
+
const currentB = b.inventory.current;
173
+
174
+
const mapA = new Map(a.data.map((item) => [item.id, item]));
175
+
const mapB = new Map(b.data.map((item) => [item.id, item]));
176
+
177
+
// Combine all known ids from both sides
178
+
const allIds = new Set([
179
+
...Object.keys(currentA),
180
+
...Object.keys(currentB),
181
+
]);
182
+
183
+
/** @type {Record<string, string>} */
184
+
const current = {};
185
+
186
+
/** @type {T[]} */
187
+
const data = [];
188
+
189
+
// Construct `current` and `data`
190
+
for await (const id of allIds) {
191
+
if (allRemoved.has(id)) continue;
192
+
193
+
if (id in currentA && id in currentB) {
194
+
const itemA = mapA.get(id);
195
+
const itemB = mapB.get(id);
196
+
197
+
if (!itemA || !itemB) {
198
+
console.warn("Should have found item but didn't!");
199
+
continue;
200
+
}
201
+
202
+
const isANewerThanB = Temporal.ZonedDateTime.compare(
203
+
Temporal.ZonedDateTime.from(itemA.updatedAt),
204
+
Temporal.ZonedDateTime.from(itemB.updatedAt),
205
+
);
206
+
207
+
const newestItem = isANewerThanB ? itemA : itemB;
208
+
const oldItem = isANewerThanB ? itemB : itemA;
209
+
210
+
/** @type {T} */
211
+
const mergedItem = { ...oldItem };
212
+
213
+
deepDiff.applyDiff(newestItem, mergedItem);
214
+
215
+
const cid = await CID.create(0x71, encode(mergedItem));
216
+
217
+
data.push(mergedItem);
218
+
current[id] = cid;
219
+
} else {
220
+
const item = mapA.get(id) ?? mapB.get(id);
221
+
222
+
if (item) {
223
+
data.push(item);
224
+
current[id] = currentA[id] ?? currentB[id];
225
+
}
226
+
}
227
+
}
228
+
229
+
// New inventory
230
+
const updatedInventory = { current, removed: Array.from(allRemoved) };
231
+
232
+
return {
233
+
cid: await CID.create(0x71, encode(updatedInventory)),
234
+
data,
235
+
inventory: updatedInventory,
236
+
};
237
+
}
238
+
239
+
/**
240
+
* @template {{ id: string }} T
241
+
* @param {{ previous: Container<T> | undefined, collection: T[] }} _
242
+
* @returns {Promise<Container<T>>}
243
+
*/
244
+
async save({ previous, collection }) {
245
+
const inventory = previous?.inventory ?? { current: {}, removed: [] };
246
+
247
+
const collIds = collection.map(({ id }) => id);
248
+
249
+
const currSet = new Set(Object.keys(inventory.current));
250
+
const collSet = new Set(collIds);
251
+
252
+
const newSet = collSet.difference(currSet);
253
+
const remSet = currSet.difference(collSet);
254
+
255
+
const alreadyRemoved = new Set(inventory.removed);
256
+
const allRemoved = alreadyRemoved.union(remSet);
257
+
258
+
/** @type {Record<string, string>} */
259
+
const current = { ...inventory.current };
260
+
261
+
/** @type Promise<void>[] */
262
+
const promises = [];
263
+
264
+
collection.forEach((a) => {
265
+
if (!newSet.has(a.id)) return;
266
+
267
+
// Item is new, calculate CID and add it to the `current` dictionary
268
+
const encoded = encode(a);
269
+
270
+
promises.push((async () => {
271
+
const cid = await CID.create(0x71, encoded);
272
+
current[a.id] = cid;
273
+
})());
274
+
});
275
+
276
+
await Promise.all(promises);
277
+
278
+
const newInventory = {
279
+
current,
280
+
removed: Array.from(allRemoved),
281
+
};
282
+
283
+
return {
284
+
// TODO: Do we need this? Too big of a perf penalty?
285
+
cid: await CID.create(0x71, encode(newInventory)),
286
+
data: collection,
287
+
inventory: newInventory,
288
+
};
289
+
}
290
+
291
+
// RENDER
292
+
293
+
/**
294
+
* @param {RenderArg} _
295
+
*/
296
+
render({ html }) {
297
+
return html`
298
+
<dop-indexed-db
299
+
namespace="${ifDefined(this.getAttribute(`namespace`))}"
300
+
></dop-indexed-db>
301
+
`;
302
+
}
303
+
}
304
+
305
+
export default DaslBytesSyncOutputTransformer;
306
+
307
+
////////////////////////////////////////////
308
+
// REGISTER
309
+
////////////////////////////////////////////
310
+
311
+
export const CLASS = DaslBytesSyncOutputTransformer;
312
+
export const NAME = "dtob-automerge";
313
+
314
+
customElements.define(NAME, CLASS);
+21
src/components/transformer/output/bytes/dasl-sync/types.d.ts
+21
src/components/transformer/output/bytes/dasl-sync/types.d.ts
···
1
+
export type Container<T> = {
2
+
/**
3
+
* CID of the inventory,
4
+
* which in turns represents the current state of the data.
5
+
*/
6
+
cid: string;
7
+
data: T[];
8
+
inventory: Inventory;
9
+
};
10
+
11
+
export type Inventory = {
12
+
/**
13
+
* `id` to `cid` map.
14
+
*/
15
+
current: Record<string, string>;
16
+
17
+
/**
18
+
* List of `id`s
19
+
*/
20
+
removed: string[];
21
+
};