Open History to see the last commit
1import { Atom, type AtomRegistry } from "effect/reactivity";
2
3import type { DraftComposerAttachment } from "../../lib/composerImages";
4import type { ComposerDraft } from "../../state/use-composer-drafts";
5import type {
6 IncomingShareDestination,
7 IncomingShareDraft,
8 selectIncomingShareAttachmentsForServer,
9} from "../sharing/incoming-share-model";
10
11export type ShareImportPhase =
12 | "idle"
13 | "reserving"
14 | "attaching"
15 | "consuming"
16 | "done"
17 | "failed"
18 | "cancelling"
19 | "cancelled";
20
21export type ShareImportEvent =
22 | "start"
23 | "reserved"
24 | "attached"
25 | "consumed"
26 | "failed"
27 | "cancel"
28 | "cancelFailed"
29 | "cancelled";
30
31const TRANSITIONS: Readonly<
32 Record<ShareImportPhase, Partial<Record<ShareImportEvent, ShareImportPhase>>>
33> = {
34 idle: { start: "reserving" },
35 reserving: { reserved: "attaching", failed: "failed" },
36 attaching: { attached: "consuming", failed: "failed" },
37 consuming: { consumed: "done", failed: "failed" },
38 done: {},
39 // "start" from failed is a retry.
40 failed: { start: "reserving", cancel: "cancelling" },
41 cancelling: { cancelled: "cancelled", cancelFailed: "failed" },
42 cancelled: {},
43};
44
45/** Events that do not apply to the current phase leave it unchanged. */
46export function transitionShareImport(
47 phase: ShareImportPhase,
48 event: ShareImportEvent,
49): ShareImportPhase {
50 return TRANSITIONS[phase][event] ?? phase;
51}
52
53function isShareImportInFlight(phase: ShareImportPhase): boolean {
54 return phase === "reserving" || phase === "attaching" || phase === "consuming";
55}
56
57/**
58 * One draft route importing one share into one draft. Plain data, compared
59 * structurally, so the screen can build it on every render.
60 */
61export interface ShareImportTarget {
62 /** The draft route's flow key; the import is abandoned once it leaves the stack. */
63 readonly flowKey: string;
64 readonly routeKey: string;
65 readonly draftKey: string;
66 readonly destination: IncomingShareDestination;
67 readonly shareId: string;
68}
69
70export interface ShareImportDeps {
71 readonly registry: AtomRegistry.AtomRegistry;
72 /** The share as the inbox holds it now. */
73 readonly getShare: (shareId: string) => IncomingShareDraft | null;
74 readonly reserve: (shareId: string, destination: IncomingShareDestination) => Promise<void>;
75 readonly release: (shareId: string, destination: IncomingShareDestination) => Promise<void>;
76 readonly consume: (shareId: string) => Promise<void>;
77 readonly readDraft: (draftKey: string) => ComposerDraft;
78 /**
79 * Restores `snapshot` if the draft still equals `merged`; otherwise removes only
80 * what the merge added, keeping later edits.
81 */
82 readonly undoMerge: (
83 draftKey: string,
84 snapshot: ComposerDraft,
85 merged: ComposerDraft,
86 ) => Promise<void>;
87 readonly mergeIntoDraft: (
88 draftKey: string,
89 content: {
90 readonly text: string;
91 readonly attachments: ReadonlyArray<DraftComposerAttachment>;
92 readonly sourceShareId: string;
93 },
94 ) => Promise<{ readonly skippedAttachmentCount: number }>;
95 readonly selectAttachments: (
96 share: IncomingShareDraft,
97 environmentId: string,
98 ) => ReturnType<typeof selectIncomingShareAttachmentsForServer>;
99 /** Sweeps files no draft retained; the sweep re-checks ownership. */
100 readonly releaseUnusedAttachments: (attachments: ReadonlyArray<DraftComposerAttachment>) => void;
101 readonly onWarnings: (warnings: ReadonlyArray<string>) => void;
102 readonly onImportFailure: (target: ShareImportTarget, error: unknown) => void;
103 /** Reported even after the route left: the undo is still incomplete. */
104 readonly onCancelFailure: (target: ShareImportTarget, error: unknown) => void;
105 /** Leaves the draft route once its import is cancelled. */
106 readonly leave: (target: ShareImportTarget) => void;
107}
108
109interface Entry {
110 readonly target: ShareImportTarget;
111 phase: ShareImportPhase;
112 readonly controller: AbortController;
113 /** Set when another draft of the same route took over before this import finished. */
114 superseded: boolean;
115 /** Set when the route left while a failed import still waits for retry or cancel. */
116 detached: boolean;
117 run: Promise<void>;
118 /** The draft before the first attempt merged into it; cancel restores it. */
119 backup: ComposerDraft | null;
120 /** The draft as the last merge published it; cancel undoes against it. */
121 merged: ComposerDraft | null;
122 reservedFor: IncomingShareDestination | null;
123 consumed: boolean;
124 needsRestore: boolean;
125 readonly listeners: Set<() => void>;
126}
127
128function keyOf(target: ShareImportTarget): string {
129 return JSON.stringify([
130 target.flowKey,
131 target.routeKey,
132 target.draftKey,
133 target.destination.environmentId,
134 target.destination.projectId,
135 target.shareId,
136 ]);
137}
138
139interface ImportedShare {
140 readonly flowKey: string;
141 readonly shareId: string;
142}
143
144/**
145 * Share imports for draft routes. Imports do durable writes, so each is tied to
146 * its target, not an atom node that can be rebuilt. Reading a target's phase atom
147 * starts its import at most once; alerts, navigation and notifications all
148 * follow an await, never run inside the read.
149 */
150export function makeIncomingShareImports(deps: ShareImportDeps) {
151 const entries = new Map<string, Entry>();
152 // Shares each route already moved into a draft. Sending clears the draft's
153 // import receipt a frame before the route leaves, so the route needs its
154 // own record to tell "imported" from "vanished".
155 const importedAtom = Atom.make<ReadonlyArray<ImportedShare>>([]).pipe(
156 Atom.keepAlive,
157 Atom.withLabel("mobile:incoming-share-imported"),
158 );
159 const isImportedAtom = Atom.family((key: string) =>
160 Atom.make((get) =>
161 get(importedAtom).some(
162 (imported) => JSON.stringify([imported.flowKey, imported.shareId]) === key,
163 ),
164 ).pipe(Atom.withLabel("mobile:incoming-share-imported:route")),
165 );
166 // One flow key per running attempt, abandoned ones included: an abandoned
167 // attempt can still be inside a draft write when the screen's target goes away.
168 const runningAtom = Atom.make<ReadonlyArray<string>>([]).pipe(
169 Atom.keepAlive,
170 Atom.withLabel("mobile:incoming-share-import-running"),
171 );
172 const isRunningAtom = Atom.family((flowKey: string) =>
173 Atom.make((get) => get(runningAtom).includes(flowKey)).pipe(
174 Atom.withLabel("mobile:incoming-share-import-running:route"),
175 ),
176 );
177
178 function notify(entry: Entry): void {
179 // Listeners rebuild their node, which re-registers them; iterate a copy.
180 for (const listener of Array.from(entry.listeners)) listener();
181 }
182
183 function advance(entry: Entry, event: ShareImportEvent): void {
184 const next = transitionShareImport(entry.phase, event);
185 if (next === entry.phase) return;
186 entry.phase = next;
187 notify(entry);
188 }
189
190 function settle(key: string, entry: Entry): void {
191 if (entry.detached && entries.get(key) === entry) entries.delete(key);
192 }
193
194 async function runImport(key: string, entry: Entry): Promise<void> {
195 const { signal } = entry.controller;
196 const { shareId, destination, draftKey } = entry.target;
197 // A phase atom node of a superseded import may still be cached; rebuilding
198 // it lets a route that switches back start over instead of showing the
199 // abandoned import's last phase.
200 const stopped = () => {
201 if (!signal.aborted) return false;
202 if (entry.superseded) notify(entry);
203 return true;
204 };
205 try {
206 await deps.reserve(shareId, destination);
207 entry.reservedFor = destination;
208 // The durable reservation makes an interrupted transfer resume only in
209 // this project instead of copying into a second project draft.
210 if (stopped()) return;
211 advance(entry, "reserved");
212 const share = deps.getShare(shareId);
213 if (!share) throw new Error("The shared content is no longer in the inbox.");
214 const selected = deps.selectAttachments(share, destination.environmentId);
215 if (selected.status === "pending") {
216 throw new Error("Server attachment support is still loading.");
217 }
218 entry.needsRestore = true;
219 let skippedAttachmentCount: number;
220 try {
221 ({ skippedAttachmentCount } = await deps.mergeIntoDraft(draftKey, {
222 text: share.text,
223 attachments: selected.attachments,
224 sourceShareId: shareId,
225 }));
226 } finally {
227 // The merge publishes before it persists, so read it back even when
228 // persisting rejects.
229 entry.merged = deps.readDraft(draftKey);
230 }
231 if (stopped()) return;
232 advance(entry, "attached");
233 await deps.consume(shareId);
234 entry.consumed = true;
235 // The consumed inbox draft was the last owner of files that never made
236 // it into the composer draft (unsupported server, oversize, limit
237 // skips). Release them even when the route left meanwhile.
238 const retainedIds = new Set(
239 deps.readDraft(draftKey).attachments.map((attachment) => attachment.id),
240 );
241 deps.releaseUnusedAttachments(
242 share.attachments.filter((attachment) => !retainedIds.has(attachment.id)),
243 );
244 deps.registry.set(importedAtom, [
245 ...deps.registry.get(importedAtom),
246 { flowKey: entry.target.flowKey, shareId },
247 ]);
248 advance(entry, "consumed");
249 entry.backup = null;
250 settle(key, entry);
251 if (stopped()) return;
252 const warnings = [...share.warnings, ...selected.warnings];
253 if (skippedAttachmentCount > 0) {
254 warnings.push(
255 `${skippedAttachmentCount} shared file${skippedAttachmentCount === 1 ? " was" : "s were"} skipped because this draft reached the attachment limit.`,
256 );
257 }
258 if (warnings.length > 0) deps.onWarnings(warnings);
259 } catch (error) {
260 if (stopped()) return;
261 advance(entry, "failed");
262 deps.onImportFailure(entry.target, error);
263 }
264 }
265
266 function begin(key: string, entry: Entry): void {
267 entry.backup ??= deps.readDraft(entry.target.draftKey);
268 advance(entry, "start");
269 const { flowKey } = entry.target;
270 deps.registry.set(runningAtom, [...deps.registry.get(runningAtom), flowKey]);
271 entry.run = runImport(key, entry).finally(() => {
272 const running = [...deps.registry.get(runningAtom)];
273 running.splice(running.indexOf(flowKey), 1);
274 deps.registry.set(runningAtom, running);
275 });
276 }
277
278 function abandon(key: string, entry: Entry): void {
279 entry.controller.abort();
280 entries.delete(key);
281 }
282
283 function start(target: ShareImportTarget): Entry {
284 const key = keyOf(target);
285 const existing = entries.get(key);
286 if (existing) return existing;
287 // A route imports into one draft at a time; a superseded import stops at
288 // its next step and no longer reports failures.
289 for (const [otherKey, other] of entries) {
290 if (other.target.flowKey === target.flowKey && isShareImportInFlight(other.phase)) {
291 other.superseded = true;
292 abandon(otherKey, other);
293 }
294 }
295 const entry: Entry = {
296 target,
297 phase: "idle",
298 controller: new AbortController(),
299 superseded: false,
300 detached: false,
301 run: Promise.resolve(),
302 backup: null,
303 merged: null,
304 reservedFor: null,
305 consumed: false,
306 needsRestore: false,
307 listeners: new Set(),
308 };
309 entries.set(key, entry);
310 begin(key, entry);
311 return entry;
312 }
313
314 async function cancel(target: ShareImportTarget): Promise<void> {
315 const key = keyOf(target);
316 const entry = entries.get(key);
317 if (!entry || entry.phase !== "failed") return;
318 const { draftKey, shareId } = entry.target;
319 advance(entry, "cancel");
320 // Cancelling finishes even if the route leaves meanwhile: it only undoes
321 // durable writes this import made.
322 try {
323 if (entry.needsRestore && entry.backup) {
324 // The draft can change while a failed import waits for an answer, by the
325 // user or an upload finishing, and a retried undo runs after its own
326 // publish. Undoing against the merged draft keeps those changes.
327 const merged = entry.merged ?? entry.backup;
328 await deps.undoMerge(draftKey, entry.backup, merged);
329 entry.needsRestore = false;
330 // Sweep the share's merged-in files only once the inbox entry was
331 // consumed: before that, the inbox still owns them for a later import.
332 // The sweep skips files a draft still holds.
333 if (entry.consumed) deps.releaseUnusedAttachments(merged.attachments);
334 }
335 if (entry.reservedFor) {
336 await deps.release(shareId, entry.reservedFor);
337 entry.reservedFor = null;
338 }
339 advance(entry, "cancelled");
340 if (entry.detached) settle(key, entry);
341 else deps.leave(entry.target);
342 } catch (error) {
343 advance(entry, "cancelFailed");
344 deps.onCancelFailure(entry.target, error);
345 }
346 }
347
348 return {
349 phaseAtom: Atom.family((target: ShareImportTarget) =>
350 Atom.make((get): ShareImportPhase => {
351 const entry = start(target);
352 const rebuild = () => get.refreshSelf();
353 entry.listeners.add(rebuild);
354 get.addFinalizer(() => entry.listeners.delete(rebuild));
355 return entry.phase;
356 }).pipe(Atom.withLabel("mobile:incoming-share-import")),
357 ),
358 /**
359 * Whether an import for this route is still running, including one whose
360 * target the screen no longer renders.
361 */
362 inFlightAtom: isRunningAtom,
363 /** Resolves once the target's current attempt stops. */
364 settled(target: ShareImportTarget): Promise<void> {
365 return entries.get(keyOf(target))?.run ?? Promise.resolve();
366 },
367 /** Whether this route already moved the share into one of its drafts. */
368 importedAtom(flowKey: string, shareId: string) {
369 return isImportedAtom(JSON.stringify([flowKey, shareId]));
370 },
371 retry(target: ShareImportTarget): void {
372 const key = keyOf(target);
373 const entry = entries.get(key);
374 if (!entry || entry.phase !== "failed") return;
375 begin(key, entry);
376 },
377 cancel,
378 /**
379 * Abandons the imports of draft routes that are no longer on the stack.
380 * A failed import keeps its retry and cancel until the user answers.
381 * Resolves once the abandoned imports have stopped.
382 */
383 prune(liveFlowKeys: ReadonlySet<string>): Promise<void> {
384 const stopping: Array<Promise<void>> = [];
385 for (const [key, entry] of entries) {
386 if (liveFlowKeys.has(entry.target.flowKey)) continue;
387 if (entry.phase === "failed" || entry.phase === "cancelling") {
388 entry.detached = true;
389 } else {
390 abandon(key, entry);
391 stopping.push(entry.run);
392 }
393 }
394 const imported = deps.registry.get(importedAtom);
395 const kept = imported.filter((share) => liveFlowKeys.has(share.flowKey));
396 if (kept.length !== imported.length) deps.registry.set(importedAtom, kept);
397 return Promise.all(stopping).then(() => undefined);
398 },
399 };
400}
401
402export const idleShareImportPhaseAtom = Atom.make<ShareImportPhase>("idle").pipe(
403 Atom.withLabel("mobile:incoming-share-import:idle"),
404);
405
406export const idleShareFlagAtom = Atom.make(false).pipe(
407 Atom.withLabel("mobile:incoming-share-flag:idle"),
408);