Form work

packages/effect-fc/src/Form.ts

@@ -71,16 +71,6 @@ export const make: {
     )
 })
 
-const run = <A, I, R>(self: Form<A, I, R>) => Stream.runForEach(
-    self.encodedValueRef.changes,
-    flow(
-        Schema.decode(self.schema, { errors: "all" }),
-        Effect.andThen(v => SubscriptionRef.set(self.valueRef, Option.some(v))),
-        Effect.andThen(SubscriptionRef.set(self.errorRef, Option.none())),
-        Effect.catchTag("ParseError", e => SubscriptionRef.set(self.errorRef, Option.some(e)))
-    ),
-)
-
 export namespace service {
     export interface Options<in out A, in out I, out R> extends make.Options<A, I, R> {}
 }
@@ -104,6 +94,16 @@ export const useForm: {
     return form
 })
 
+const run = <A, I, R>(self: Form<A, I, R>) => Stream.runForEach(
+    self.encodedValueRef.changes,
+    flow(
+        Schema.decode(self.schema, { errors: "all" }),
+        Effect.andThen(v => SubscriptionRef.set(self.valueRef, Option.some(v))),
+        Effect.andThen(SubscriptionRef.set(self.errorRef, Option.none())),
+        Effect.catchTag("ParseError", e => SubscriptionRef.set(self.errorRef, Option.some(e)))
+    ),
+)
+
 
 export namespace useInput {
     export interface Options {
@@ -139,18 +139,31 @@ export const useInput: {
     const [value, setValue] = yield* Hooks.useRefState(internalValueRef)
     const [issues] = yield* Hooks.useSubscribables(issuesSubscribable)
 
-    yield* Hooks.useFork(() => Stream.runForEach(
-        internalValueRef.changes.pipe(
-            Stream.changesWith(Equivalence.strict()),
-            options?.debounce ? Stream.debounce(options.debounce) : identity,
-            Stream.drop(1),
+    yield* Hooks.useFork(() => Effect.all([
+        Stream.runForEach(
+            self.encodedValueRef.changes.pipe(
+                Stream.flatMap(PropertyPath.get(path)),
+                Stream.drop(1),
+            ),
+
+            upstreamEncodedValue => internalValueRef.pipe(
+
+            ),
         ),
 
-        internalValue => self.encodedValueRef.pipe(
-            Effect.andThen(encodedValue => PropertyPath.immutableSet(encodedValue, path, internalValue)),
-            Effect.andThen(encodedValue => SubscriptionRef.set(self.encodedValueRef, encodedValue)),
+        Stream.runForEach(
+            internalValueRef.changes.pipe(
+                Stream.changesWith(Equivalence.strict()),
+                options?.debounce ? Stream.debounce(options.debounce) : identity,
+                Stream.drop(1),
+            ),
+
+            internalValue => self.encodedValueRef.pipe(
+                Effect.andThen(encodedValue => PropertyPath.immutableSet(encodedValue, path, internalValue)),
+                Effect.andThen(encodedValue => SubscriptionRef.set(self.encodedValueRef, encodedValue)),
+            ),
         ),
-    ), [internalValueRef, self, ...path])
+    ], { concurrency: "unbounded" }), [internalValueRef, self, ...path])
 
     return { value, setValue, issues }
 })