1/** Construct a streaming push from an application's own protocol or background job. */
2import * as Effect from "effect/Effect";
3import * as Fiber from "effect/Fiber";
4import * as Schema from "effect/Schema";
5import * as Stream from "effect/Stream";
6import { Oid, RefName } from "./Api/Schema.ts";
7import { StoreError } from "./Protocol/Store.ts";
8import { incomingStates, type PushInput, type RefUpdate } from "./Push.ts";
9import { makeStreamingSource } from "./Store/StreamingSource.ts";
10
11const Commands = Schema.Array(
12 Schema.Struct({ ref: RefName, oldOid: Oid, newOid: Oid }),
13);
14
15/**
16 * Prepare input from decoded ref updates and raw pack bytes, with no HTTP dependency.
17 * Use Stream.empty for a ref-only mutation. The input and producer share the caller's scope.
18 */
19export const fromStream = <E, R>(
20 updates: ReadonlyArray<RefUpdate>,
21 pack: Stream.Stream<Uint8Array, E, R>,
22 options?: { readonly atomic?: boolean; readonly declaredBytes?: number },
23) =>
24 Effect.gen(function* () {
25 const commands = yield* Schema.decodeUnknownEffect(Commands)(updates);
26 const input: PushInput = Object.freeze({
27 updates: Object.freeze(
28 commands.map((command) => Object.freeze({ ...command })),
29 ),
30 atomic: options?.atomic ?? true,
31 });
32 const feeder = makeStreamingSource();
33 let total = 0;
34 const receiving = yield* Effect.forkScoped(
35 pack.pipe(
36 Stream.runForEach((bytes) =>
37 Effect.andThen(
38 feeder.push(bytes),
39 Effect.sync(() => {
40 total += bytes.length;
41 }),
42 ),
43 ),
44 Effect.map(() => {
45 feeder.end();
46 return { total };
47 }),
48 Effect.mapError(
49 (error) =>
50 new StoreError({ reason: `pack stream failed: ${String(error)}` }),
51 ),
52 Effect.tapError((error) => Effect.sync(() => feeder.fail(error))),
53 Effect.result,
54 ),
55 );
56 const state = {
57 feeder,
58 receiving,
59 packStart: 0,
60 declaredBytes: options?.declaredBytes,
61 active: true,
62 claimed: false,
63 };
64 incomingStates.set(input, state);
65 yield* Effect.addFinalizer(() =>
66 Effect.gen(function* () {
67 state.active = false;
68 feeder.fail(new StoreError({ reason: "push scope closed" }));
69 yield* Fiber.interrupt(receiving);
70 }),
71 );
72 return input;
73 });